-#!/usr/bin/env python
+#!/usr/bin/env python3
"""L2XC Multi-instance Test Case HLD:
**NOTES:**
- - higher number (more than 15) of pg-l2 interfaces causes problems => only \
- 14 pg-l2 interfaces and 10 cross-connects are tested
- - jumbo packets in configuration with 14 l2-pg interfaces leads to \
- problems too
+ - higher number (more than 15) of pg-l2 interfaces causes problems => only
+ 14 pg-l2 interfaces and 10 cross-connects are tested
+ - jumbo packets in configuration with 14 l2-pg interfaces leads to
+ problems too
**config 1**
- add 14 pg-l2 interfaces
- send L2 MAC frames between all pairs of pg-l2 interfaces
**verify 1**
- - all packets received correctly in case of cross-connected l2-pg interfaces
+ - all packets received correctly in case of cross-connected l2-pg
+ interfaces
- no packet received in case of not cross-connected l2-pg interfaces
**config 2**
- send L2 MAC frames between all pairs of pg-l2 interfaces
**verify 2**
- - all packets received correctly in case of cross-connected l2-pg interfaces
+ - all packets received correctly in case of cross-connected l2-pg
+ interfaces
- no packet received in case of not cross-connected l2-pg interfaces
**config 3**
- send L2 MAC frames between all pairs of pg-l2 interfaces
**verify 3**
- - all packets received correctly in case of cross-connected l2-pg interfaces
+ - all packets received correctly in case of cross-connected l2-pg
+ interfaces
- no packet received in case of not cross-connected l2-pg interfaces
**config 4**
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
-from framework import VppTestCase, VppTestRunner
-from util import Host
+from framework import VppTestCase
+from asfframework import VppTestRunner
+from util import Host, ppp
class TestL2xcMultiInst(VppTestCase):
- """ L2XC Multi-instance Test Case """
+ """L2XC Multi-instance Test Case"""
@classmethod
def setUpClass(cls):
cls.flows = dict()
for i in range(len(cls.pg_interfaces)):
delta = 1 if i % 2 == 0 else -1
- cls.flows[cls.pg_interfaces[i]] = [cls.pg_interfaces[i+delta]]
+ cls.flows[cls.pg_interfaces[i]] = [cls.pg_interfaces[i + delta]]
# Mapping between packet-generator index and lists of test hosts
cls.hosts_by_pg_idx = dict()
super(TestL2xcMultiInst, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestL2xcMultiInst, cls).tearDownClass()
+
def setUp(self):
"""
Clear trace and packet infos before running each test.
"""
super(TestL2xcMultiInst, self).setUp()
- self.packet_infos = {}
+ self.reset_packet_infos()
def tearDown(self):
"""
Show various debug prints after each test.
"""
super(TestL2xcMultiInst, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.ppcli("show l2patch"))
+
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.ppcli("show l2patch"))
@classmethod
def create_hosts(cls, count):
:param int count: Number of hosts to create MAC/IPv4 addresses for.
"""
n_int = len(cls.pg_interfaces)
- macs_per_if = count / n_int
+ macs_per_if = count // n_int
i = -1
for pg_if in cls.pg_interfaces:
i += 1
for j in range(start_nr, end_nr):
host = Host(
"00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
- "172.17.1%02u.%u" % (pg_if.sw_if_index, j))
+ "172.17.1%02u.%u" % (pg_if.sw_if_index, j),
+ )
hosts.append(host)
def create_xconnects(self, count, start=0):
(Default value = 0)
"""
for i in range(count):
- rx_if = self.pg_interfaces[i+start]
+ rx_if = self.pg_interfaces[i + start]
delta = 1 if i % 2 == 0 else -1
- tx_if = self.pg_interfaces[i+start+delta]
- self.vapi.sw_interface_set_l2_xconnect(rx_if.sw_if_index,
- tx_if.sw_if_index, 1)
- self.logger.info("Cross-connect from %s to %s created"
- % (tx_if.name, rx_if.name))
+ tx_if = self.pg_interfaces[i + start + delta]
+ self.vapi.sw_interface_set_l2_xconnect(
+ rx_if.sw_if_index, tx_if.sw_if_index, 1
+ )
+ self.logger.info(
+ "Cross-connect from %s to %s created" % (tx_if.name, rx_if.name)
+ )
if self.pg_in_xc.count(rx_if) == 0:
self.pg_in_xc.append(rx_if)
if self.pg_not_in_xc.count(rx_if) == 1:
(Default value = 0)
"""
for i in range(count):
- rx_if = self.pg_interfaces[i+start]
+ rx_if = self.pg_interfaces[i + start]
delta = 1 if i % 2 == 0 else -1
- tx_if = self.pg_interfaces[i+start+delta]
- self.vapi.sw_interface_set_l2_xconnect(rx_if.sw_if_index,
- tx_if.sw_if_index, 0)
- self.logger.info("Cross-connect from %s to %s deleted"
- % (tx_if.name, rx_if.name))
+ tx_if = self.pg_interfaces[i + start + delta]
+ self.vapi.sw_interface_set_l2_xconnect(
+ rx_if.sw_if_index, tx_if.sw_if_index, 0
+ )
+ self.logger.info(
+ "Cross-connect from %s to %s deleted" % (tx_if.name, rx_if.name)
+ )
if self.pg_not_in_xc.count(rx_if) == 0:
self.pg_not_in_xc.append(rx_if)
if self.pg_in_xc.count(rx_if) == 1:
for i in range(0, n_int):
dst_host = dst_hosts[i]
src_host = random.choice(src_hosts)
- pkt_info = self.create_packet_info(
- src_if.sw_if_index, dst_if.sw_if_index)
+ pkt_info = self.create_packet_info(src_if, dst_if)
payload = self.info_to_payload(pkt_info)
- p = (Ether(dst=dst_host.mac, src=src_host.mac) /
- IP(src=src_host.ip4, dst=dst_host.ip4) /
- UDP(sport=1234, dport=1234) /
- Raw(payload))
+ p = (
+ Ether(dst=dst_host.mac, src=src_host.mac)
+ / IP(src=src_host.ip4, dst=dst_host.ip4)
+ / UDP(sport=1234, dport=1234)
+ / Raw(payload)
+ )
pkt_info.data = p.copy()
size = random.choice(packet_sizes)
self.extend_packet(p, size)
pkts.append(p)
- self.logger.debug("Input stream created for port %s. Length: %u pkt(s)"
- % (src_if.name, len(pkts)))
+ self.logger.debug(
+ "Input stream created for port %s. Length: %u pkt(s)"
+ % (src_if.name, len(pkts))
+ )
return pkts
def verify_capture(self, pg_if, capture):
last_info[i.sw_if_index] = None
dst_sw_if_index = pg_if.sw_if_index
for packet in capture:
- payload_info = self.payload_to_info(str(packet[Raw]))
+ payload_info = self.payload_to_info(packet[Raw])
try:
ip = packet[IP]
udp = packet[UDP]
packet_index = payload_info.index
self.assertEqual(payload_info.dst, dst_sw_if_index)
- self.logger.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])
+ payload_info.src, dst_sw_if_index, last_info[payload_info.src]
+ )
last_info[payload_info.src] = next_info
self.assertTrue(next_info is not None)
self.assertEqual(packet_index, next_info.index)
self.assertEqual(udp.sport, saved_packet[UDP].sport)
self.assertEqual(udp.dport, saved_packet[UDP].dport)
except:
- self.logger.error("Unexpected or invalid packet:")
- self.logger.error(packet.show())
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
for i in self.pg_interfaces:
remaining_packet = self.get_next_packet_info_for_interface2(
- i, dst_sw_if_index, last_info[i.sw_if_index])
+ i, dst_sw_if_index, last_info[i.sw_if_index]
+ )
self.assertTrue(
remaining_packet is None,
- "Port %u: Packet expected from source %u didn't arrive" %
- (dst_sw_if_index, i.sw_if_index))
+ "Port %u: Packet expected from source %u didn't arrive"
+ % (dst_sw_if_index, i.sw_if_index),
+ )
def run_verify_test(self):
"""
- Create packet streams for all configured l2-pg interfaces, send all
+ Create packet streams for all configured l2-pg interfaces, send all \
prepared packet streams and verify that:
- - all packets received correctly on all pg-l2 interfaces assigned \
- to cross-connects
- - no packet received on all pg-l2 interfaces not assigned to \
- cross-connects
-
- :raise: RuntimeError if no packet captured on l2-pg interface assigned \
- to the cross-connect or if any packet is captured on l2-pg interface \
- not assigned to the cross-connect.
+ - all packets received correctly on all pg-l2 interfaces assigned
+ to cross-connects
+ - no packet received on all pg-l2 interfaces not assigned to
+ cross-connects
+
+ :raise RuntimeError: if no packet captured on l2-pg interface assigned
+ to the cross-connect or if any packet is captured
+ on l2-pg interface not assigned to the
+ cross-connect.
"""
# Test
# Create incoming packet streams for packet-generator interfaces
# Verify
# Verify outgoing packet streams per packet-generator interface
for pg_if in self.pg_interfaces:
- capture = pg_if.get_capture()
if pg_if in self.pg_in_xc:
- if len(capture) == 0:
- raise RuntimeError("Interface %s is cross-connect sink but "
- "the capture is empty!" % pg_if.name)
+ capture = pg_if.get_capture(remark="interface is a cross-connect sink")
self.verify_capture(pg_if, capture)
elif pg_if in self.pg_not_in_xc:
- try:
- self.assertEqual(len(capture), 0)
- except AssertionError:
- raise RuntimeError("Interface %s is not cross-connect sink "
- "but the capture is not empty!"
- % pg_if.name)
+ pg_if.assert_nothing_captured(
+ remark="interface is not a cross-connect sink"
+ )
else:
- self.logger.error("Unknown interface: %s" % pg_if.name)
+ raise Exception("Unexpected interface: %s" % pg_if.name)
def test_l2xc_inst_01(self):
- """ L2XC Multi-instance test 1 - create 10 cross-connects
- """
+ """L2XC Multi-instance test 1 - create 10 cross-connects"""
# Config 1
# Create 10 cross-connects
self.create_xconnects(10)
self.run_verify_test()
def test_l2xc_inst_02(self):
- """ L2XC Multi-instance test 2 - delete 4 cross-connects
- """
+ """L2XC Multi-instance test 2 - delete 4 cross-connects"""
# Config 2
# Delete 4 cross-connects
self.delete_xconnects(4)
self.run_verify_test()
def test_l2xc_inst_03(self):
- """ L2BD Multi-instance 3 - add new 4 cross-connects
- """
+ """L2BD Multi-instance 3 - add new 4 cross-connects"""
# Config 3
# Add new 4 cross-connects
self.create_xconnects(4, start=10)
self.run_verify_test()
def test_l2xc_inst_04(self):
- """ L2XC Multi-instance test 4 - delete 10 cross-connects
- """
+ """L2XC Multi-instance test 4 - delete 10 cross-connects"""
# Config 4
# Delete 10 cross-connects
self.delete_xconnects(10, start=4)
self.run_verify_test()
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff