X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_span.py;h=3572d64e79c44e31ebc59c30f8460dce53bd5a69;hb=5c801b362a536fcae704c50bf1573362d372bb3c;hp=dddd57f9b8fa9ce9b168eb098ec001fb56a8567a;hpb=e63325e3ca03c847963863446345e6c80a2c0cfd;p=vpp.git diff --git a/test/test_span.py b/test/test_span.py index dddd57f9b8f..3572d64e79c 100644 --- a/test/test_span.py +++ b/test/test_span.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 import unittest @@ -9,24 +9,26 @@ from scapy.layers.vxlan import VXLAN from framework import VppTestCase, VppTestRunner from util import Host, ppp -from vpp_sub_interface import VppDot1QSubint, VppDot1ADSubint -from vpp_gre_interface import VppGreInterface, VppGre6Interface -from vpp_papi_provider import L2_VTR_OP +from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint, VppDot1ADSubint +from vpp_gre_interface import VppGreInterface +from vpp_vxlan_tunnel import VppVxlanTunnel from collections import namedtuple +from vpp_papi import VppEnum -Tag = namedtuple('Tag', ['dot1', 'vlan']) + +Tag = namedtuple("Tag", ["dot1", "vlan"]) DOT1AD = 0x88A8 DOT1Q = 0x8100 class TestSpan(VppTestCase): - """ SPAN Test Case """ + """SPAN Test Case""" @classmethod def setUpClass(cls): super(TestSpan, cls).setUpClass() # Test variables - cls.pkts_per_burst = 257 # Number of packets per burst + cls.pkts_per_burst = 257 # Number of packets per burst # create 3 pg interfaces cls.create_pg_interfaces(range(3)) @@ -52,26 +54,28 @@ class TestSpan(VppTestCase): i.config_ip4() i.resolve_arp() - cls.vxlan = cls.vapi.vxlan_add_del_tunnel( - src_address=cls.pg2.local_ip4n, dst_address=cls.pg2.remote_ip4n, - is_add=1, vni=1111) - def setUp(self): super(TestSpan, self).setUp() + self.vxlan = VppVxlanTunnel( + self, src=self.pg2.local_ip4, dst=self.pg2.remote_ip4, vni=1111 + ) + self.vxlan.add_vpp_config() self.reset_packet_infos() def tearDown(self): super(TestSpan, self).tearDown() - if not self.vpp_dead: - self.logger.info(self.vapi.ppcli("show interface span")) + + def show_commands_at_teardown(self): + self.logger.info(self.vapi.ppcli("show interface span")) def xconnect(self, a, b, is_add=1): self.vapi.sw_interface_set_l2_xconnect(a, b, enable=is_add) self.vapi.sw_interface_set_l2_xconnect(b, a, enable=is_add) def bridge(self, sw_if_index, is_add=1): - self.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=sw_if_index, - bd_id=self.bd_id, enable=is_add) + self.vapi.sw_interface_set_l2_bridge( + rx_sw_if_index=sw_if_index, bd_id=self.bd_id, enable=is_add + ) def _remove_tag(self, packet, vlan, tag_type): self.assertEqual(packet.type, tag_type) @@ -142,11 +146,13 @@ class TestSpan(VppTestCase): for i in range(0, self.pkts_per_burst): payload = "span test" - size = packet_sizes[(i / 2) % len(packet_sizes)] - p = (Ether(src=src_if.local_mac, dst=dst_mac) / - IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) / - UDP(sport=10000 + src_if.sw_if_index * 1000 + i, dport=1234) / - Raw(payload)) + size = packet_sizes[int((i / 2) % len(packet_sizes))] + p = ( + Ether(src=src_if.local_mac, dst=dst_mac) + / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) + / UDP(sport=10000 + src_if.sw_if_index * 1000 + i, dport=1234) + / Raw(payload) + ) if do_dot1: p = self.sub_if.add_dot1_layer(p) self.extend_packet(p, size) @@ -154,9 +160,12 @@ class TestSpan(VppTestCase): return pkts def verify_capture(self, cap1, cap2): - self.assertEqual(len(cap1), len(cap2), - "Different number of sent and mirrored packets :" - "%u != %u" % (len(cap1), len(cap2))) + self.assertEqual( + len(cap1), + len(cap2), + "Different number of sent and mirrored packets :" + "%u != %u" % (len(cap1), len(cap2)), + ) pkts1 = [(pkt[Ether] / pkt[IP] / pkt[UDP]) for pkt in cap1] pkts2 = [(pkt[Ether] / pkt[IP] / pkt[UDP]) for pkt in cap2] @@ -164,7 +173,7 @@ class TestSpan(VppTestCase): self.assertEqual(pkts1.sort(), pkts2.sort()) def test_device_span(self): - """ SPAN device rx mirror """ + """SPAN device rx mirror""" # Create bi-directional cross-connects between pg0 and pg1 self.xconnect(self.pg0.sw_if_index, self.pg1.sw_if_index) @@ -174,7 +183,8 @@ class TestSpan(VppTestCase): # Enable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.pg0.sw_if_index, self.pg2.sw_if_index) + self.pg0.sw_if_index, self.pg2.sw_if_index + ) self.logger.info(self.vapi.ppcli("show interface span")) # Enable packet capturing and start packet sending @@ -188,13 +198,14 @@ class TestSpan(VppTestCase): # Disable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.pg0.sw_if_index, self.pg2.sw_if_index, state=0) + self.pg0.sw_if_index, self.pg2.sw_if_index, state=0 + ) self.xconnect(self.pg0.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_pkts) def test_span_l2_rx(self): - """ SPAN l2 rx mirror """ + """SPAN l2 rx mirror""" self.sub_if.admin_up() @@ -202,13 +213,13 @@ class TestSpan(VppTestCase): # Create bi-directional cross-connects between pg0 subif and pg1 self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index) # Create incoming packet streams for packet-generator interfaces - pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pkts) # Enable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1) + self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1 + ) self.logger.info(self.vapi.ppcli("show interface span")) # Enable packet capturing and start packet sending @@ -223,29 +234,29 @@ class TestSpan(VppTestCase): # Disable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1) + self.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1 + ) self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_pkts) def test_span_l2_rx_dst_vxlan(self): - """ SPAN l2 rx mirror into vxlan """ + """SPAN l2 rx mirror into vxlan""" self.sub_if.admin_up() - self.vapi.sw_interface_set_flags(self.vxlan.sw_if_index, - flags=1) + self.vapi.sw_interface_set_flags(self.vxlan.sw_if_index, flags=1) self.bridge(self.vxlan.sw_if_index, is_add=1) # Create bi-directional cross-connects between pg0 subif and pg1 self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index) # Create incoming packet streams for packet-generator interfaces - pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pkts) # Enable SPAN on pg0 sub if (mirrored to vxlan) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.vxlan.sw_if_index, is_l2=1) + self.sub_if.sw_if_index, self.vxlan.sw_if_index, is_l2=1 + ) self.logger.info(self.vapi.ppcli("show interface span")) # Enable packet capturing and start packet sending @@ -260,19 +271,23 @@ class TestSpan(VppTestCase): self.bridge(self.vxlan.sw_if_index, is_add=0) # Disable SPAN on pg0 sub if (mirrored to vxlan) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.vxlan.sw_if_index, state=0, is_l2=1) + self.sub_if.sw_if_index, self.vxlan.sw_if_index, state=0, is_l2=1 + ) self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_pkts) def test_span_l2_rx_dst_gre_erspan(self): - """ SPAN l2 rx mirror into gre-erspan """ + """SPAN l2 rx mirror into gre-erspan""" self.sub_if.admin_up() - gre_if = VppGreInterface(self, self.pg2.local_ip4, - self.pg2.remote_ip4, - type=2, - session=543) + gre_if = VppGreInterface( + self, + self.pg2.local_ip4, + self.pg2.remote_ip4, + session=543, + type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_ERSPAN), + ) gre_if.add_vpp_config() gre_if.admin_up() @@ -282,13 +297,13 @@ class TestSpan(VppTestCase): self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1) # Create incoming packet streams for packet-generator interfaces - pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pkts) # Enable SPAN on pg0 sub if (mirrored to gre-erspan) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, gre_if.sw_if_index, is_l2=1) + self.sub_if.sw_if_index, gre_if.sw_if_index, is_l2=1 + ) # Enable packet capturing and start packet sending self.pg_enable_capture(self.pg_interfaces) @@ -299,27 +314,33 @@ class TestSpan(VppTestCase): pg1_pkts = self.pg1.get_capture(n_pkts) pg2_pkts = self.pg2.get_capture(n_pkts) - def decap(p): return self.decap_erspan(p, session=543) + def decap(p): + return self.decap_erspan(p, session=543) + pg2_decaped = [decap(p) for p in pg2_pkts] self.bridge(gre_if.sw_if_index, is_add=0) # Disable SPAN on pg0 sub if self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, gre_if.sw_if_index, state=0, is_l2=1) + self.sub_if.sw_if_index, gre_if.sw_if_index, state=0, is_l2=1 + ) gre_if.remove_vpp_config() self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_decaped) def test_span_l2_rx_dst_gre_subif_vtr(self): - """ SPAN l2 rx mirror into gre-subif+vtr """ + """SPAN l2 rx mirror into gre-subif+vtr""" self.sub_if.admin_up() - gre_if = VppGreInterface(self, self.pg2.local_ip4, - self.pg2.remote_ip4, - type=1) + gre_if = VppGreInterface( + self, + self.pg2.local_ip4, + self.pg2.remote_ip4, + type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB), + ) gre_if.add_vpp_config() gre_if.admin_up() @@ -333,13 +354,13 @@ class TestSpan(VppTestCase): self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1) # Create incoming packet streams for packet-generator interfaces - pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pkts) # Enable SPAN on pg0 sub if (mirrored to gre sub if) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, gre_sub_if.sw_if_index, is_l2=1) + self.sub_if.sw_if_index, gre_sub_if.sw_if_index, is_l2=1 + ) # Enable packet capturing and start packet sending self.pg_enable_capture(self.pg_interfaces) @@ -350,22 +371,24 @@ class TestSpan(VppTestCase): pg1_pkts = self.pg1.get_capture(n_pkts) pg2_pkts = self.pg2.get_capture(n_pkts) - def decap(p): return self.remove_tags( - self.decap_gre(p), [Tag(dot1=DOT1Q, vlan=500)]) + def decap(p): + return self.remove_tags(self.decap_gre(p), [Tag(dot1=DOT1Q, vlan=500)]) + pg2_decaped = [decap(p) for p in pg2_pkts] self.bridge(gre_sub_if.sw_if_index, is_add=0) # Disable SPAN on pg0 sub if self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, gre_sub_if.sw_if_index, state=0, is_l2=1) + self.sub_if.sw_if_index, gre_sub_if.sw_if_index, state=0, is_l2=1 + ) gre_if.remove_vpp_config() self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_decaped) def test_span_l2_rx_dst_1q_vtr(self): - """ SPAN l2 rx mirror into 1q subif+vtr """ + """SPAN l2 rx mirror into 1q subif+vtr""" self.sub_if.admin_up() self.vlan_sub_if.admin_up() @@ -375,12 +398,12 @@ class TestSpan(VppTestCase): self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1) # Create incoming packet streams for packet-generator interfaces - pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pkts) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.vlan_sub_if.sw_if_index, is_l2=1) + self.sub_if.sw_if_index, self.vlan_sub_if.sw_if_index, is_l2=1 + ) # Enable packet capturing and start packet sending self.pg_enable_capture(self.pg_interfaces) @@ -390,20 +413,21 @@ class TestSpan(VppTestCase): n_pkts = len(pkts) pg1_pkts = self.pg1.get_capture(n_pkts) pg2_pkts = self.pg2.get_capture(n_pkts) - pg2_untagged = [self.remove_tags(p, [Tag(dot1=DOT1Q, vlan=300)]) - for p in pg2_pkts] + pg2_untagged = [ + self.remove_tags(p, [Tag(dot1=DOT1Q, vlan=300)]) for p in pg2_pkts + ] self.bridge(self.vlan_sub_if.sw_if_index, is_add=0) # Disable SPAN on pg0 sub if (mirrored to vxlan) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.vlan_sub_if.sw_if_index, state=0, - is_l2=1) + self.sub_if.sw_if_index, self.vlan_sub_if.sw_if_index, state=0, is_l2=1 + ) self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_untagged) def test_span_l2_rx_dst_1ad_vtr(self): - """ SPAN l2 rx mirror into 1ad subif+vtr """ + """SPAN l2 rx mirror into 1ad subif+vtr""" self.sub_if.admin_up() self.qinq_sub_if.admin_up() @@ -413,12 +437,12 @@ class TestSpan(VppTestCase): self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1) # Create incoming packet streams for packet-generator interfaces - pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pkts) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.qinq_sub_if.sw_if_index, is_l2=1) + self.sub_if.sw_if_index, self.qinq_sub_if.sw_if_index, is_l2=1 + ) # Enable packet capturing and start packet sending self.pg_enable_capture(self.pg_interfaces) @@ -428,34 +452,35 @@ class TestSpan(VppTestCase): n_pkts = len(pkts) pg1_pkts = self.pg1.get_capture(n_pkts) pg2_pkts = self.pg2.get_capture(n_pkts) - pg2_untagged = [self.remove_tags(p, [Tag(dot1=DOT1AD, vlan=400), - Tag(dot1=DOT1Q, vlan=500)]) - for p in pg2_pkts] + pg2_untagged = [ + self.remove_tags(p, [Tag(dot1=DOT1AD, vlan=400), Tag(dot1=DOT1Q, vlan=500)]) + for p in pg2_pkts + ] self.bridge(self.qinq_sub_if.sw_if_index, is_add=0) # Disable SPAN on pg0 sub if (mirrored to vxlan) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.qinq_sub_if.sw_if_index, state=0, - is_l2=1) + self.sub_if.sw_if_index, self.qinq_sub_if.sw_if_index, state=0, is_l2=1 + ) self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_untagged) def test_l2_tx_span(self): - """ SPAN l2 tx mirror """ + """SPAN l2 tx mirror""" self.sub_if.admin_up() self.bridge(self.pg2.sw_if_index) # Create bi-directional cross-connects between pg0 and pg1 self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index) # Create incoming packet streams for packet-generator interfaces - pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pkts) # Enable SPAN on pg1 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.pg1.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=2) + self.pg1.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=2 + ) self.logger.info(self.vapi.ppcli("show interface span")) # Enable packet capturing and start packet sending @@ -469,13 +494,14 @@ class TestSpan(VppTestCase): self.bridge(self.pg2.sw_if_index, is_add=0) # Disable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.pg1.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1) + self.pg1.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1 + ) self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg1_pkts, pg2_pkts) def test_l2_rx_tx_span(self): - """ SPAN l2 rx tx mirror """ + """SPAN l2 rx tx mirror""" self.sub_if.admin_up() self.bridge(self.pg2.sw_if_index) @@ -483,16 +509,15 @@ class TestSpan(VppTestCase): self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index) # Create incoming packet streams for packet-generator interfaces - pg0_pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True) + pg0_pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True) self.pg0.add_stream(pg0_pkts) - pg1_pkts = self.create_stream( - self.pg1, self.pg_if_packet_sizes, do_dot1=False) + pg1_pkts = self.create_stream(self.pg1, self.pg_if_packet_sizes, do_dot1=False) self.pg1.add_stream(pg1_pkts) # Enable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=3) + self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=3 + ) self.logger.info(self.vapi.ppcli("show interface span")) # Enable packet capturing and start packet sending @@ -511,34 +536,40 @@ class TestSpan(VppTestCase): self.bridge(self.pg2.sw_if_index, is_add=0) # Disable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1) + self.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1 + ) self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0) self.verify_capture(pg0_pkts + pg1_pkts, pg2_pkts) def test_l2_bcast_mirror(self): - """ SPAN l2 broadcast mirror """ + """SPAN l2 broadcast mirror""" self.sub_if.admin_up() self.bridge(self.pg2.sw_if_index) # Create bi-directional cross-connects between pg0 and pg1 self.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=1) + rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=1 + ) self.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=1) + rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=1 + ) # Create incoming packet streams for packet-generator interfaces pg0_pkts = self.create_stream( - self.pg0, self.pg_if_packet_sizes, do_dot1=True, bcast=True) + self.pg0, self.pg_if_packet_sizes, do_dot1=True, bcast=True + ) self.pg0.add_stream(pg0_pkts) pg1_pkts = self.create_stream( - self.pg1, self.pg_if_packet_sizes, do_dot1=False, bcast=True) + self.pg1, self.pg_if_packet_sizes, do_dot1=False, bcast=True + ) self.pg1.add_stream(pg1_pkts) # Enable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=3) + self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=3 + ) self.logger.info(self.vapi.ppcli("show interface span")) # Enable packet capturing and start packet sending @@ -556,15 +587,18 @@ class TestSpan(VppTestCase): self.bridge(self.pg2.sw_if_index, is_add=0) self.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=0) + rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=0 + ) self.vapi.sw_interface_set_l2_bridge( - rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=0) + rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=0 + ) # Disable SPAN on pg0 (mirrored to pg2) self.vapi.sw_interface_span_enable_disable( - self.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1) + self.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1 + ) self.verify_capture(pg0_pkts + pg1_pkts, pg2_pkts) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)