Add single-loop test variant to L2BD and L2XC tests
- create single-loop version of tests
- update doc strings
- add possibility to log CLI string for logging level "info"
Change-Id: Ibc9e5650b8a33d2ed168a8440c2ae03227be60bb
Signed-off-by: Jan <jgelety@cisco.com>
diff --git a/test/doc/modules.rst b/test/doc/modules.rst
index 6d456d7..773283e 100644
--- a/test/doc/modules.rst
+++ b/test/doc/modules.rst
@@ -10,7 +10,7 @@
run_tests
scapy_handlers
template_bd
- test_ip
+ test_ip4
test_ip6
test_l2bd
test_l2xc
diff --git a/test/doc/test_ip.rst b/test/doc/test_ip4.rst
similarity index 77%
rename from test/doc/test_ip.rst
rename to test/doc/test_ip4.rst
index a1e97b8..c3c3a23 100644
--- a/test/doc/test_ip.rst
+++ b/test/doc/test_ip4.rst
@@ -1,7 +1,7 @@
test_ip module
==============
-.. automodule:: test_ip
+.. automodule:: test_ip4
:members:
:undoc-members:
:show-inheritance:
diff --git a/test/framework.py b/test/framework.py
index 8dbc18f..fdb600c 100644
--- a/test/framework.py
+++ b/test/framework.py
@@ -276,11 +276,11 @@
def tearDown(self):
""" Show various debug prints after each test """
if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show int"))
- self.logger.info(self.vapi.cli("show trace"))
- self.logger.info(self.vapi.cli("show hardware"))
- self.logger.info(self.vapi.cli("show error"))
- self.logger.info(self.vapi.cli("show run"))
+ self.logger.info(self.vapi.ppcli("show int"))
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.logger.info(self.vapi.ppcli("show hardware"))
+ self.logger.info(self.vapi.ppcli("show error"))
+ self.logger.info(self.vapi.ppcli("show run"))
def setUp(self):
""" Clear trace before running each test"""
diff --git a/test/test_l2bd.py b/test/test_l2bd.py
index 3c65cc1..46ba2e4 100644
--- a/test/test_l2bd.py
+++ b/test/test_l2bd.py
@@ -1,7 +1,6 @@
#!/usr/bin/env python
import unittest
-from logging import *
import random
from scapy.packet import Raw
@@ -9,78 +8,111 @@
from scapy.layers.inet import IP, UDP
from framework import VppTestCase, VppTestRunner
-from vpp_sub_interface import VppDot1QSubint
from util import Host
+from vpp_sub_interface import VppDot1QSubint, VppDot1ADSubint
class TestL2bd(VppTestCase):
""" L2BD Test Case """
- # Test variables
- bd_id = 1 # Bridge domain ID
- mac_entries_count = 100 # Number of MAC entries for bridge-domain to learn
- dot1q_sub_id = 100 # SubID of dot1q sub-interface
- dot1q_tag = 100 # VLAN tag for dot1q sub-interface
- dot1ad_sub_id = 200 # SubID of dot1ad sub-interface
- dot1ad_outer_tag = 200 # VLAN S-tag for dot1ad sub-interface
- dot1ad_inner_tag = 300 # VLAN C-tag for dot1ad sub-interface
- pkts_per_burst = 257 # Number of packets per burst
-
@classmethod
def setUpClass(cls):
+ """
+ Perform standard class setup (defined by class method setUpClass in
+ class VppTestCase) before running the test case, set test case related
+ variables and configure VPP.
+
+ :var int bd_id: Bridge domain ID.
+ :var int mac_entries_count: Number of MAC entries for bridge-domain to
+ learn.
+ :var int dot1q_tag: VLAN tag for dot1q sub-interface.
+ :var int dot1ad_sub_id: SubID of dot1ad sub-interface.
+ :var int dot1ad_outer_tag: VLAN S-tag for dot1ad sub-interface.
+ :var int dot1ad_inner_tag: VLAN C-tag for dot1ad sub-interface.
+ :var int sl_pkts_per_burst: Number of packets in burst for single-loop
+ test.
+ :var int dl_pkts_per_burst: Number of packets in burst for dual-loop
+ test.
+ """
super(TestL2bd, cls).setUpClass()
+ # Test variables
+ cls.bd_id = 1
+ cls.mac_entries_count = 100
+ # cls.dot1q_sub_id = 100
+ cls.dot1q_tag = 100
+ cls.dot1ad_sub_id = 20
+ cls.dot1ad_outer_tag = 200
+ cls.dot1ad_inner_tag = 300
+ cls.sl_pkts_per_burst = 2
+ cls.dl_pkts_per_burst = 257
+
+ try:
+ # create 3 pg interfaces
+ cls.create_pg_interfaces(range(3))
+
+ # create 2 sub-interfaces for pg1 and pg2
+ cls.sub_interfaces = [
+ VppDot1QSubint(cls, cls.pg1, cls.dot1q_tag),
+ VppDot1ADSubint(cls, cls.pg2, cls.dot1ad_sub_id,
+ cls.dot1ad_outer_tag, cls.dot1ad_inner_tag)]
+
+ # packet flows mapping pg0 -> pg1, pg2, etc.
+ cls.flows = dict()
+ cls.flows[cls.pg0] = [cls.pg1, cls.pg2]
+ cls.flows[cls.pg1] = [cls.pg0, cls.pg2]
+ cls.flows[cls.pg2] = [cls.pg0, cls.pg1]
+
+ # packet sizes
+ cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
+ cls.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
+
+ cls.interfaces = list(cls.pg_interfaces)
+ cls.interfaces.extend(cls.sub_interfaces)
+
+ # Create BD with MAC learning enabled and put interfaces and
+ # sub-interfaces to this BD
+ for pg_if in cls.pg_interfaces:
+ sw_if_index = pg_if.sub_if.sw_if_index \
+ if hasattr(pg_if, 'sub_if') else pg_if.sw_if_index
+ cls.vapi.sw_interface_set_l2_bridge(sw_if_index,
+ bd_id=cls.bd_id)
+
+ # setup all interfaces
+ for i in cls.interfaces:
+ i.admin_up()
+
+ # mapping between packet-generator index and lists of test hosts
+ cls.hosts_by_pg_idx = dict()
+
+ # create test host entries and inject packets to learn MAC entries
+ # in the bridge-domain
+ cls.create_hosts_and_learn(cls.mac_entries_count)
+ cls.logger.info(cls.vapi.ppcli("show l2fib"))
+
+ except Exception:
+ super(TestL2bd, cls).tearDownClass()
+ raise
+
def setUp(self):
+ """
+ Clear trace and packet infos before running each test.
+ """
super(TestL2bd, self).setUp()
-
- # create 3 pg interfaces
- self.create_pg_interfaces(range(3))
-
- # create 2 sub-interfaces for pg1 and pg2
- self.sub_interfaces = [
- VppDot1QSubint(self, self.pg1, TestL2bd.dot1q_sub_id),
- VppDot1QSubint(self, self.pg2, TestL2bd.dot1ad_sub_id)]
-
- # packet flows mapping pg0 -> pg1, pg2, etc.
- self.flows = dict()
- self.flows[self.pg0] = [self.pg1, self.pg2]
- self.flows[self.pg1] = [self.pg0, self.pg2]
- self.flows[self.pg2] = [self.pg0, self.pg1]
-
- # packet sizes
- self.pg_if_packet_sizes = [64, 512, 1518, 9018]
- self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
-
- self.interfaces = list(self.pg_interfaces)
- self.interfaces.extend(self.sub_interfaces)
-
- # Create BD with MAC learning enabled and put interfaces and
- # sub-interfaces to this BD
- for pg_if in self.pg_interfaces:
- sw_if_index = pg_if.sub_if.sw_if_index if hasattr(pg_if, 'sub_if') \
- else pg_if.sw_if_index
- self.vapi.sw_interface_set_l2_bridge(sw_if_index,
- bd_id=TestL2bd.bd_id)
-
- # setup all interfaces
- for i in self.interfaces:
- i.admin_up()
-
- # mapping between packet-generator index and lists of test hosts
- self.hosts_by_pg_idx = dict()
-
- # create test host entries and inject packets to learn MAC entries in
- # the bridge-domain
- self.create_hosts_and_learn(TestL2bd.mac_entries_count)
- info(self.vapi.cli("show l2fib"))
+ self.packet_infos = {}
def tearDown(self):
+ """
+ Show various debug prints after each test.
+ """
super(TestL2bd, self).tearDown()
if not self.vpp_dead:
- info(self.vapi.cli("show l2fib verbose"))
- info(self.vapi.cli("show bridge-domain %s detail" % self.bd_id))
+ self.logger.info(self.vapi.ppcli("show l2fib verbose"))
+ self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" %
+ self.bd_id))
- def create_hosts_and_learn(self, count):
+ @classmethod
+ def create_hosts_and_learn(cls, count):
"""
Create required number of host MAC addresses and distribute them among
interfaces. Create host IPv4 address for every host MAC address. Create
@@ -89,15 +121,15 @@
:param count: Integer number of hosts to create MAC/IPv4 addresses for.
"""
- n_int = len(self.pg_interfaces)
+ n_int = len(cls.pg_interfaces)
macs_per_if = count / n_int
i = -1
- for pg_if in self.pg_interfaces:
+ for pg_if in cls.pg_interfaces:
i += 1
start_nr = macs_per_if * i
end_nr = count if i == (n_int - 1) else macs_per_if * (i + 1)
- self.hosts_by_pg_idx[pg_if.sw_if_index] = []
- hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
+ cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
+ hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
packets = []
for j in range(start_nr, end_nr):
host = Host(
@@ -109,12 +141,20 @@
packet = pg_if.sub_if.add_dot1_layer(packet)
packets.append(packet)
pg_if.add_stream(packets)
- info("Sending broadcast eth frames for MAC learning")
- self.pg_start()
+ cls.logger.info("Sending broadcast eth frames for MAC learning")
+ cls.pg_start()
- def create_stream(self, src_if, packet_sizes):
+ def create_stream(self, src_if, packet_sizes, packets_per_burst):
+ """
+ Create input packet stream for defined interface.
+
+ :param object src_if: Interface to create packet stream for.
+ :param list packet_sizes: List of required packet sizes.
+ :param int packets_per_burst: Number of packets in burst.
+ :return: Stream of packets.
+ """
pkts = []
- for i in range(0, TestL2bd.pkts_per_burst):
+ for i in range(0, packets_per_burst):
dst_if = self.flows[src_if][i % 2]
dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
@@ -128,12 +168,18 @@
pkt_info.data = p.copy()
if hasattr(src_if, 'sub_if'):
p = src_if.sub_if.add_dot1_layer(p)
- size = packet_sizes[(i / 2) % len(packet_sizes)]
+ size = random.choice(packet_sizes)
self.extend_packet(p, size)
pkts.append(p)
return pkts
def verify_capture(self, pg_if, capture):
+ """
+ Verify captured input packet stream for defined interface.
+
+ :param object pg_if: Interface to verify captured packet stream for.
+ :param list capture: Captured packet stream.
+ """
last_info = dict()
for i in self.pg_interfaces:
last_info[i.sw_if_index] = None
@@ -156,8 +202,8 @@
udp = packet[UDP]
packet_index = payload_info.index
self.assertEqual(payload_info.dst, dst_sw_if_index)
- debug("Got packet on port %s: src=%u (id=%u)" %
- (pg_if.name, payload_info.src, packet_index))
+ self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
+ (pg_if.name, payload_info.src, packet_index))
next_info = self.get_next_packet_info_for_interface2(
payload_info.src, dst_sw_if_index,
last_info[payload_info.src])
@@ -171,8 +217,8 @@
self.assertEqual(udp.sport, saved_packet[UDP].sport)
self.assertEqual(udp.dport, saved_packet[UDP].dport)
except:
- error("Unexpected or invalid packet:")
- error(packet.show())
+ self.logger.error("Unexpected or invalid packet:")
+ self.logger.error(packet.show())
raise
for i in self.pg_interfaces:
remaining_packet = self.get_next_packet_info_for_interface2(
@@ -182,25 +228,14 @@
"Port %u: Packet expected from source %u didn't arrive" %
(dst_sw_if_index, i.sw_if_index))
- def test_l2bd(self):
- """ L2BD MAC learning test
-
- 1.config
- MAC learning enabled
- learn 100 MAC enries
- 3 interfaces: untagged, dot1q, dot1ad (dot1q used instead of dot1ad
- in the first version)
-
- 2.sending l2 eth pkts between 3 interface
- 64B, 512B, 1518B, 9200B (ether_size)
- burst of 257 pkts per interface
- """
+ def run_l2bd_test(self, pkts_per_burst):
+ """ L2BD MAC learning test """
# Create incoming packet streams for packet-generator interfaces
for i in self.pg_interfaces:
packet_sizes = self.sub_if_packet_sizes if hasattr(i, 'sub_if') \
else self.pg_if_packet_sizes
- pkts = self.create_stream(i, packet_sizes)
+ pkts = self.create_stream(i, packet_sizes, pkts_per_burst)
i.add_stream(pkts)
# Enable packet capture and start packet sending
@@ -210,9 +245,43 @@
# Verify outgoing packet streams per packet-generator interface
for i in self.pg_interfaces:
capture = i.get_capture()
- info("Verifying capture on interface %s" % i.name)
+ self.logger.info("Verifying capture on interface %s" % i.name)
self.verify_capture(i, capture)
+ def test_l2bd_sl(self):
+ """ L2BD MAC learning single-loop test
+
+ Test scenario:
+ 1.config
+ MAC learning enabled
+ learn 100 MAC enries
+ 3 interfaces: untagged, dot1q, dot1ad (dot1q used instead of
+ dot1ad in the first version)
+
+ 2.sending l2 eth pkts between 3 interface
+ 64B, 512B, 1518B, 9200B (ether_size)
+ burst of 2 pkts per interface
+ """
+
+ self.run_l2bd_test(self.sl_pkts_per_burst)
+
+ def test_l2bd_dl(self):
+ """ L2BD MAC learning dual-loop test
+
+ Test scenario:
+ 1.config
+ MAC learning enabled
+ learn 100 MAC enries
+ 3 interfaces: untagged, dot1q, dot1ad (dot1q used instead of
+ dot1ad in the first version)
+
+ 2.sending l2 eth pkts between 3 interface
+ 64B, 512B, 1518B, 9200B (ether_size)
+ burst of 257 pkts per interface
+ """
+
+ self.run_l2bd_test(self.dl_pkts_per_burst)
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
diff --git a/test/test_l2xc.py b/test/test_l2xc.py
index 23fd757..49ca996 100644
--- a/test/test_l2xc.py
+++ b/test/test_l2xc.py
@@ -6,7 +6,6 @@
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
-from logging import *
from framework import VppTestCase, VppTestRunner
from util import Host
@@ -15,84 +14,112 @@
class TestL2xc(VppTestCase):
""" L2XC Test Case """
- # Test variables
- hosts_nr = 10 # Number of hosts
- pkts_per_burst = 257 # Number of packets per burst
-
@classmethod
def setUpClass(cls):
+ """
+ Perform standard class setup (defined by class method setUpClass in
+ class VppTestCase) before running the test case, set test case related
+ variables and configure VPP.
+
+ :var int hosts_nr: Number of hosts to be created.
+ :var int dl_pkts_per_burst: Number of packets in burst for dual-loop
+ test.
+ :var int sl_pkts_per_burst: Number of packets in burst for single-loop
+ test.
+ """
super(TestL2xc, cls).setUpClass()
+ # Test variables
+ cls.hosts_nr = 10
+ cls.dl_pkts_per_burst = 257
+ cls.sl_pkts_per_burst = 2
+
+ try:
+ # create 4 pg interfaces
+ cls.create_pg_interfaces(range(4))
+
+ # packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
+ cls.flows = dict()
+ cls.flows[cls.pg0] = [cls.pg1]
+ cls.flows[cls.pg1] = [cls.pg0]
+ cls.flows[cls.pg2] = [cls.pg3]
+ cls.flows[cls.pg3] = [cls.pg2]
+
+ # packet sizes
+ cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
+
+ cls.interfaces = list(cls.pg_interfaces)
+
+ # Create bi-directional cross-connects between pg0 and pg1
+ cls.vapi.sw_interface_set_l2_xconnect(
+ cls.pg0.sw_if_index, cls.pg1.sw_if_index, enable=1)
+ cls.vapi.sw_interface_set_l2_xconnect(
+ cls.pg1.sw_if_index, cls.pg0.sw_if_index, enable=1)
+
+ # Create bi-directional cross-connects between pg2 and pg3
+ cls.vapi.sw_interface_set_l2_xconnect(
+ cls.pg2.sw_if_index, cls.pg3.sw_if_index, enable=1)
+ cls.vapi.sw_interface_set_l2_xconnect(
+ cls.pg3.sw_if_index, cls.pg2.sw_if_index, enable=1)
+
+ # mapping between packet-generator index and lists of test hosts
+ cls.hosts_by_pg_idx = dict()
+
+ # Create host MAC and IPv4 lists
+ cls.create_host_lists(cls.hosts_nr)
+
+ # setup all interfaces
+ for i in cls.interfaces:
+ i.admin_up()
+
+ except Exception:
+ super(TestL2xc, cls).tearDownClass()
+ raise
+
def setUp(self):
+ """
+ Clear trace and packet infos before running each test.
+ """
super(TestL2xc, self).setUp()
-
- # create 4 pg interfaces
- self.create_pg_interfaces(range(4))
-
- # packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
- self.flows = dict()
- self.flows[self.pg0] = [self.pg1]
- self.flows[self.pg1] = [self.pg0]
- self.flows[self.pg2] = [self.pg3]
- self.flows[self.pg3] = [self.pg2]
-
- # packet sizes
- self.pg_if_packet_sizes = [64, 512, 1518, 9018]
-
- self.interfaces = list(self.pg_interfaces)
-
- # Create bi-directional cross-connects between pg0 and pg1
- self.vapi.sw_interface_set_l2_xconnect(
- self.pg0.sw_if_index, self.pg1.sw_if_index, enable=1)
- self.vapi.sw_interface_set_l2_xconnect(
- self.pg1.sw_if_index, self.pg0.sw_if_index, enable=1)
-
- # Create bi-directional cross-connects between pg2 and pg3
- self.vapi.sw_interface_set_l2_xconnect(
- self.pg2.sw_if_index, self.pg3.sw_if_index, enable=1)
- self.vapi.sw_interface_set_l2_xconnect(
- self.pg3.sw_if_index, self.pg2.sw_if_index, enable=1)
-
- info(self.vapi.cli("show l2patch"))
-
- # mapping between packet-generator index and lists of test hosts
- self.hosts_by_pg_idx = dict()
-
- # Create host MAC and IPv4 lists
- # self.MY_MACS = dict()
- # self.MY_IP4S = dict()
- self.create_host_lists(TestL2xc.hosts_nr)
-
- # setup all interfaces
- for i in self.interfaces:
- i.admin_up()
+ self.packet_infos = {}
def tearDown(self):
+ """
+ Show various debug prints after each test.
+ """
super(TestL2xc, self).tearDown()
if not self.vpp_dead:
- info(self.vapi.cli("show l2patch"))
+ self.logger.info(self.vapi.ppcli("show l2patch"))
- def create_host_lists(self, count):
- """ Method to create required number of MAC and IPv4 addresses.
+ @classmethod
+ def create_host_lists(cls, count):
+ """
+ Method to create required number of MAC and IPv4 addresses.
Create required number of host MAC addresses and distribute them among
interfaces. Create host IPv4 address for every host MAC address too.
:param count: Number of hosts to create MAC and IPv4 addresses for.
"""
- for pg_if in self.pg_interfaces:
- # self.MY_MACS[i.sw_if_index] = []
- # self.MY_IP4S[i.sw_if_index] = []
- self.hosts_by_pg_idx[pg_if.sw_if_index] = []
- hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
+ for pg_if in cls.pg_interfaces:
+ cls.hosts_by_pg_idx[pg_if.sw_if_index] = []
+ hosts = cls.hosts_by_pg_idx[pg_if.sw_if_index]
for j in range(0, count):
host = Host(
"00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
"172.17.1%02x.%u" % (pg_if.sw_if_index, j))
hosts.append(host)
- def create_stream(self, src_if, packet_sizes):
+ def create_stream(self, src_if, packet_sizes, packets_per_burst):
+ """
+ Create input packet stream for defined interface.
+
+ :param object src_if: Interface to create packet stream for.
+ :param list packet_sizes: List of required packet sizes.
+ :param int packets_per_burst: Number of packets in burst.
+ :return: Stream of packets.
+ """
pkts = []
- for i in range(0, TestL2xc.pkts_per_burst):
+ for i in range(0, packets_per_burst):
dst_if = self.flows[src_if][0]
dst_host = random.choice(self.hosts_by_pg_idx[dst_if.sw_if_index])
src_host = random.choice(self.hosts_by_pg_idx[src_if.sw_if_index])
@@ -104,12 +131,18 @@
UDP(sport=1234, dport=1234) /
Raw(payload))
pkt_info.data = p.copy()
- size = packet_sizes[(i / 2) % len(packet_sizes)]
+ size = random.choice(packet_sizes)
self.extend_packet(p, size)
pkts.append(p)
return pkts
def verify_capture(self, pg_if, capture):
+ """
+ Verify captured input packet stream for defined interface.
+
+ :param object pg_if: Interface to verify captured packet stream for.
+ :param list capture: Captured packet stream.
+ """
last_info = dict()
for i in self.interfaces:
last_info[i.sw_if_index] = None
@@ -121,8 +154,8 @@
payload_info = self.payload_to_info(str(packet[Raw]))
packet_index = payload_info.index
self.assertEqual(payload_info.dst, dst_sw_if_index)
- debug("Got packet on port %s: src=%u (id=%u)" %
- (pg_if.name, payload_info.src, packet_index))
+ self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
+ (pg_if.name, payload_info.src, packet_index))
next_info = self.get_next_packet_info_for_interface2(
payload_info.src, dst_sw_if_index,
last_info[payload_info.src])
@@ -136,8 +169,8 @@
self.assertEqual(udp.sport, saved_packet[UDP].sport)
self.assertEqual(udp.dport, saved_packet[UDP].dport)
except:
- error("Unexpected or invalid packet:")
- packet.show()
+ self.logger.error("Unexpected or invalid packet:")
+ self.logger.error(packet.show())
raise
for i in self.interfaces:
remaining_packet = self.get_next_packet_info_for_interface2(
@@ -146,20 +179,13 @@
"Port %u: Packet expected from source %u didn't"
" arrive" % (dst_sw_if_index, i.sw_if_index))
- def test_l2xc(self):
- """ L2XC test
-
- Test scenario:
- 1. config
- 2 pairs of 2 interfaces, l2xconnected
- 2. sending l2 eth packets between 4 interfaces
- 64B, 512B, 1518B, 9018B (ether_size)
- burst of packets per interface
- """
+ def run_l2xc_test(self, pkts_per_burst):
+ """ L2XC test """
# Create incoming packet streams for packet-generator interfaces
for i in self.interfaces:
- pkts = self.create_stream(i, self.pg_if_packet_sizes)
+ pkts = self.create_stream(i, self.pg_if_packet_sizes,
+ pkts_per_burst)
i.add_stream(pkts)
# Enable packet capturing and start packet sending
@@ -169,9 +195,37 @@
# Verify outgoing packet streams per packet-generator interface
for i in self.pg_interfaces:
capture = i.get_capture()
- info("Verifying capture on interface %s" % i.name)
+ self.logger.info("Verifying capture on interface %s" % i.name)
self.verify_capture(i, capture)
+ def test_l2xc_sl(self):
+ """ L2XC single-loop test
+
+ Test scenario:
+ 1. config
+ 2 pairs of 2 interfaces, l2xconnected
+
+ 2. sending l2 eth packets between 4 interfaces
+ 64B, 512B, 1518B, 9018B (ether_size)
+ burst of 2 packets per interface
+ """
+
+ self.run_l2xc_test(self.sl_pkts_per_burst)
+
+ def test_l2xc_dl(self):
+ """ L2XC dual-loop test
+
+ Test scenario:
+ 1. config
+ 2 pairs of 2 interfaces, l2xconnected
+
+ 2. sending l2 eth packets between 4 interfaces
+ 64B, 512B, 1518B, 9018B (ether_size)
+ burst of 257 packets per interface
+ """
+
+ self.run_l2xc_test(self.dl_pkts_per_burst)
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
diff --git a/test/vpp_papi_provider.py b/test/vpp_papi_provider.py
index 10445de..5721b6b 100644
--- a/test/vpp_papi_provider.py
+++ b/test/vpp_papi_provider.py
@@ -69,7 +69,8 @@
return reply
def cli(self, cli):
- """Execute a CLI, calling the before/after hooks appropriately
+ """
+ Execute a CLI, calling the before/after hooks appropriately.
:param cli: CLI to execute
:returns: CLI output
@@ -79,9 +80,18 @@
cli += '\n'
r = vpp_papi.cli_inband(len(cli), cli)
self.hook.after_cli(cli)
- if(hasattr(r, 'reply')):
+ if hasattr(r, 'reply'):
return r.reply[0].decode().rstrip('\x00')
+ def ppcli(self, cli):
+ """
+ Helping method to print CLI command in case of info logging level.
+
+ :param cli: CLI to execute
+ :returns: CLI output
+ """
+ return cli + "\n" + self.cli(cli)
+
def show_version(self):
""" """
return vpp_papi.show_version()