X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_l2xc.py;h=eba349a2a0f56cfe106bcb826d54bc0d5c25a39c;hb=1a9dc75fe8099fdde9b1dd248a8fca35b001f9fc;hp=2ec4af9288ee348e5c3ce588f5a9bfdf7b020c6c;hpb=dab231a11ec96e829b22ff80c612333edc5a93e6;p=vpp.git diff --git a/test/test_l2xc.py b/test/test_l2xc.py index 2ec4af9288e..eba349a2a0f 100644 --- a/test/test_l2xc.py +++ b/test/test_l2xc.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 import unittest import random @@ -12,7 +12,7 @@ from util import Host, ppp class TestL2xc(VppTestCase): - """ L2XC Test Case """ + """L2XC Test Case""" @classmethod def setUpClass(cls): @@ -52,15 +52,19 @@ class TestL2xc(VppTestCase): # 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.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) + 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.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) + 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() @@ -76,6 +80,10 @@ class TestL2xc(VppTestCase): super(TestL2xc, cls).tearDownClass() raise + @classmethod + def tearDownClass(cls): + super(TestL2xc, cls).tearDownClass() + def setUp(self): super(TestL2xc, self).setUp() self.reset_packet_infos() @@ -85,8 +93,9 @@ class TestL2xc(VppTestCase): Show various debug prints after each test. """ super(TestL2xc, 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_host_lists(cls, count): @@ -103,7 +112,8 @@ class TestL2xc(VppTestCase): 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)) + "172.17.1%02x.%u" % (pg_if.sw_if_index, j), + ) hosts.append(host) def create_stream(self, src_if, packet_sizes, packets_per_burst): @@ -122,10 +132,12 @@ class TestL2xc(VppTestCase): src_host = random.choice(self.hosts_by_pg_idx[src_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) @@ -147,14 +159,16 @@ class TestL2xc(VppTestCase): try: ip = packet[IP] udp = packet[UDP] - payload_info = self.payload_to_info(str(packet[Raw])) + payload_info = self.payload_to_info(packet[Raw]) 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) @@ -169,18 +183,20 @@ class TestL2xc(VppTestCase): raise for i in self.interfaces: remaining_packet = self.get_next_packet_info_for_interface2( - 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)) + 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), + ) def run_l2xc_test(self, pkts_per_burst): - """ L2XC test """ + """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_per_burst) + pkts = self.create_stream(i, self.pg_if_packet_sizes, pkts_per_burst) i.add_stream(pkts) # Enable packet capturing and start packet sending @@ -194,7 +210,7 @@ class TestL2xc(VppTestCase): self.verify_capture(i, capture) def test_l2xc_sl(self): - """ L2XC single-loop test + """L2XC single-loop test Test scenario: 1. config @@ -208,7 +224,7 @@ class TestL2xc(VppTestCase): self.run_l2xc_test(self.sl_pkts_per_burst) def test_l2xc_dl(self): - """ L2XC dual-loop test + """L2XC dual-loop test Test scenario: 1. config @@ -222,5 +238,5 @@ class TestL2xc(VppTestCase): self.run_l2xc_test(self.dl_pkts_per_burst) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)