summaryrefslogtreecommitdiffstats
path: root/test/asf/test_vlib.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/asf/test_vlib.py')
-rw-r--r--test/asf/test_vlib.py327
1 files changed, 0 insertions, 327 deletions
diff --git a/test/asf/test_vlib.py b/test/asf/test_vlib.py
deleted file mode 100644
index dce08b823b4..00000000000
--- a/test/asf/test_vlib.py
+++ /dev/null
@@ -1,327 +0,0 @@
-#!/usr/bin/env python3
-
-import unittest
-import pexpect
-import time
-import signal
-from config import config
-from asfframework import VppTestCase, VppTestRunner
-from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
-from scapy.layers.inet import IP, ICMP
-from scapy.layers.l2 import Ether
-from scapy.packet import Raw
-
-
-@unittest.skipUnless(config.gcov, "part of code coverage tests")
-class TestVlib(VppTestCase):
- """Vlib Unit Test Cases"""
-
- vpp_worker_count = 1
-
- @classmethod
- def setUpClass(cls):
- super(TestVlib, cls).setUpClass()
-
- @classmethod
- def tearDownClass(cls):
- super(TestVlib, cls).tearDownClass()
-
- def setUp(self):
- super(TestVlib, self).setUp()
-
- def tearDown(self):
- super(TestVlib, self).tearDown()
-
- def test_vlib_main_unittest(self):
- """Vlib main.c Code Coverage Test"""
-
- cmds = [
- "loopback create",
- "packet-generator new {\n"
- " name vlib\n"
- " limit 15\n"
- " size 128-128\n"
- " interface loop0\n"
- " node ethernet-input\n"
- " data {\n"
- " IP6: 00:d0:2d:5e:86:85 -> 00:0d:ea:d0:00:00\n"
- " ICMP: db00::1 -> db00::2\n"
- " incrementing 30\n"
- " }\n"
- "}\n",
- "event-logger trace dispatch",
- "event-logger stop",
- "event-logger clear",
- "event-logger resize 102400",
- "event-logger restart",
- "pcap dispatch trace on max 100 buffer-trace pg-input 15",
- "pa en",
- "show event-log 100 all",
- "event-log save",
- "event-log save foo",
- "pcap dispatch trace",
- "pcap dispatch trace status",
- "pcap dispatch trace off",
- "show vlib frame-allocation",
- ]
-
- for cmd in cmds:
- r = self.vapi.cli_return_response(cmd)
- if r.retval != 0:
- if hasattr(r, "reply"):
- self.logger.info(cmd + " FAIL reply " + r.reply)
- else:
- self.logger.info(cmd + " FAIL retval " + str(r.retval))
-
- def test_vlib_node_cli_unittest(self):
- """Vlib node_cli.c Code Coverage Test"""
-
- cmds = [
- "loopback create",
- "packet-generator new {\n"
- " name vlib\n"
- " limit 15\n"
- " size 128-128\n"
- " interface loop0\n"
- " node ethernet-input\n"
- " data {\n"
- " IP6: 00:d0:2d:5e:86:85 -> 00:0d:ea:d0:00:00\n"
- " ICMP: db00::1 -> db00::2\n"
- " incrementing 30\n"
- " }\n"
- "}\n",
- "show vlib graph",
- "show vlib graph ethernet-input",
- "show vlib graphviz",
- "show vlib graphviz graphviz.dot",
- "pa en",
- "show runtime ethernet-input",
- "show runtime brief verbose max summary",
- "clear runtime",
- "show node index 1",
- "show node ethernet-input",
- "show node pg-input",
- "set node function",
- "set node function no-such-node",
- "set node function cdp-input default",
- "set node function ethernet-input default",
- "set node function ethernet-input bozo",
- "set node function ethernet-input",
- "show \t",
- ]
-
- for cmd in cmds:
- r = self.vapi.cli_return_response(cmd)
- if r.retval != 0:
- if hasattr(r, "reply"):
- self.logger.info(cmd + " FAIL reply " + r.reply)
- else:
- self.logger.info(cmd + " FAIL retval " + str(r.retval))
-
- def test_vlib_buffer_c_unittest(self):
- """Vlib buffer.c Code Coverage Test"""
-
- cmds = [
- "loopback create",
- "packet-generator new {\n"
- " name vlib\n"
- " limit 15\n"
- " size 128-128\n"
- " interface loop0\n"
- " node ethernet-input\n"
- " data {\n"
- " IP6: 00:d0:2d:5e:86:85 -> 00:0d:ea:d0:00:00\n"
- " ICMP: db00::1 -> db00::2\n"
- " incrementing 30\n"
- " }\n"
- "}\n",
- "event-logger trace",
- "event-logger trace enable",
- "event-logger trace api cli barrier",
- "pa en",
- "show interface bogus",
- "event-logger trace disable api cli barrier",
- "event-logger trace circuit-node ethernet-input",
- "event-logger trace circuit-node ethernet-input disable",
- "clear interfaces",
- "test vlib",
- "test vlib2",
- "show memory api-segment stats-segment main-heap verbose",
- "leak-check { show memory }",
- "show cpu",
- "memory-trace main-heap",
- "memory-trace main-heap api-segment stats-segment",
- "leak-check { show version }",
- "show version ?",
- "comment { show version }",
- "uncomment { show version }",
- "show memory main-heap",
- "show memory bogus",
- "choices",
- "test heap-validate",
- "memory-trace main-heap disable",
- "show buffers",
- "show eve",
- "show help",
- "show ip ",
- ]
-
- for cmd in cmds:
- r = self.vapi.cli_return_response(cmd)
- if r.retval != 0:
- if hasattr(r, "reply"):
- self.logger.info(cmd + " FAIL reply " + r.reply)
- else:
- self.logger.info(cmd + " FAIL retval " + str(r.retval))
-
- def test_vlib_format_unittest(self):
- """Vlib format.c Code Coverage Test"""
-
- cmds = [
- "loopback create",
- "classify filter pcap mask l2 proto match l2 proto 0x86dd",
- "classify filter pcap del",
- "test format-vlib",
- ]
-
- for cmd in cmds:
- r = self.vapi.cli_return_response(cmd)
- if r.retval != 0:
- if hasattr(r, "reply"):
- self.logger.info(cmd + " FAIL reply " + r.reply)
- else:
- self.logger.info(cmd + " FAIL retval " + str(r.retval))
-
- def test_vlib_main_unittest(self):
- """Private Binary API Segment Test (takes 70 seconds)"""
-
- vat_path = config.vpp + "_api_test"
- vat = pexpect.spawn(vat_path, ["socket-name", self.get_api_sock_path()])
- vat.expect("vat# ", timeout=10)
- vat.sendline("sock_init_shm")
- vat.expect("vat# ", timeout=10)
- vat.sendline("sh api cli")
- vat.kill(signal.SIGKILL)
- vat.wait()
- self.logger.info("vat terminated, 70 second wait for the Reaper")
- time.sleep(70)
- self.logger.info("Reaper should be complete...")
-
- def test_pool(self):
- """Fixed-size Pool Test"""
-
- cmds = [
- "test pool",
- ]
-
- for cmd in cmds:
- r = self.vapi.cli_return_response(cmd)
- if r.retval != 0:
- if hasattr(r, "reply"):
- self.logger.info(cmd + " FAIL reply " + r.reply)
- else:
- self.logger.info(cmd + " FAIL retval " + str(r.retval))
-
-
-class TestVlibFrameLeak(VppTestCase):
- """Vlib Frame Leak Test Cases"""
-
- vpp_worker_count = 1
-
- @classmethod
- def setUpClass(cls):
- super(TestVlibFrameLeak, cls).setUpClass()
-
- @classmethod
- def tearDownClass(cls):
- super(TestVlibFrameLeak, cls).tearDownClass()
-
- def setUp(self):
- super(TestVlibFrameLeak, self).setUp()
- # create 1 pg interface
- self.create_pg_interfaces(range(1))
-
- for i in self.pg_interfaces:
- i.admin_up()
- i.config_ip4()
- i.resolve_arp()
-
- def tearDown(self):
- super(TestVlibFrameLeak, self).tearDown()
- for i in self.pg_interfaces:
- i.unconfig_ip4()
- i.admin_down()
-
- def test_vlib_mw_refork_frame_leak(self):
- """Vlib worker thread refork leak test case"""
- icmp_id = 0xB
- icmp_seq = 5
- icmp_load = b"\x0a" * 18
- pkt = (
- Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
- / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
- / ICMP(id=icmp_id, seq=icmp_seq)
- / Raw(load=icmp_load)
- )
-
- # Send a packet
- self.pg0.add_stream(pkt)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- rx = self.pg0.get_capture(1)
-
- self.assertEquals(len(rx), 1)
- rx = rx[0]
- ether = rx[Ether]
- ipv4 = rx[IP]
-
- self.assertEqual(ether.src, self.pg0.local_mac)
- self.assertEqual(ether.dst, self.pg0.remote_mac)
-
- self.assertEqual(ipv4.src, self.pg0.local_ip4)
- self.assertEqual(ipv4.dst, self.pg0.remote_ip4)
-
- # Save allocated frame count
- frame_allocated = {}
- for fs in self.vapi.cli("show vlib frame-allocation").splitlines()[1:]:
- spl = fs.split()
- thread = int(spl[0])
- size = int(spl[1])
- alloc = int(spl[2])
- key = (thread, size)
- frame_allocated[key] = alloc
-
- # cause reforks
- _ = self.create_loopback_interfaces(1)
-
- # send the same packet
- self.pg0.add_stream(pkt)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- rx = self.pg0.get_capture(1)
-
- self.assertEquals(len(rx), 1)
- rx = rx[0]
- ether = rx[Ether]
- ipv4 = rx[IP]
-
- self.assertEqual(ether.src, self.pg0.local_mac)
- self.assertEqual(ether.dst, self.pg0.remote_mac)
-
- self.assertEqual(ipv4.src, self.pg0.local_ip4)
- self.assertEqual(ipv4.dst, self.pg0.remote_ip4)
-
- # Check that no frame were leaked during refork
- for fs in self.vapi.cli("show vlib frame-allocation").splitlines()[1:]:
- spl = fs.split()
- thread = int(spl[0])
- size = int(spl[1])
- alloc = int(spl[2])
- key = (thread, size)
- self.assertEqual(frame_allocated[key], alloc)
-
-
-if __name__ == "__main__":
- unittest.main(testRunner=VppTestRunner)