from framework import tag_fixme_vpp_workers
from framework import VppTestCase, VppTestRunner
-from vpp_igmp import find_igmp_state, IGMP_FILTER, IgmpRecord, IGMP_MODE, \
- IgmpSG, VppHostState, wait_for_igmp_event
+from vpp_igmp import (
+ find_igmp_state,
+ IGMP_FILTER,
+ IgmpRecord,
+ IGMP_MODE,
+ IgmpSG,
+ VppHostState,
+ wait_for_igmp_event,
+)
from vpp_ip_route import find_mroute, VppIpTable
@tag_fixme_vpp_workers
class TestIgmp(VppTestCase):
- """ IGMP Test Case """
+ """IGMP Test Case"""
@classmethod
def setUpClass(cls):
self.pg_start()
def test_igmp_flush(self):
- """ IGMP Link Up/down and Flush """
+ """IGMP Link Up/down and Flush"""
#
# FIX THIS. Link down.
#
def test_igmp_enable(self):
- """ IGMP enable/disable on an interface
+ """IGMP enable/disable on an interface
- check for the addition/removal of the IGMP mroutes """
+ check for the addition/removal of the IGMP mroutes"""
self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 1, IGMP_MODE.HOST)
self.vapi.igmp_enable_disable(self.pg1.sw_if_index, 1, IGMP_MODE.HOST)
self.vapi.igmp_enable_disable(self.pg2.sw_if_index, 1, IGMP_MODE.HOST)
self.vapi.igmp_enable_disable(self.pg3.sw_if_index, 1, IGMP_MODE.HOST)
- self.assertTrue(find_mroute(self, "224.0.0.1", "0.0.0.0", 32,
- table_id=1))
- self.assertTrue(find_mroute(self, "224.0.0.22", "0.0.0.0", 32,
- table_id=1))
+ self.assertTrue(find_mroute(self, "224.0.0.1", "0.0.0.0", 32, table_id=1))
+ self.assertTrue(find_mroute(self, "224.0.0.22", "0.0.0.0", 32, table_id=1))
self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 0, IGMP_MODE.HOST)
self.vapi.igmp_enable_disable(self.pg1.sw_if_index, 0, IGMP_MODE.HOST)
self.vapi.igmp_enable_disable(self.pg2.sw_if_index, 0, IGMP_MODE.HOST)
self.assertTrue(find_mroute(self, "224.0.0.1", "0.0.0.0", 32))
self.assertFalse(find_mroute(self, "224.0.0.22", "0.0.0.0", 32))
- self.assertTrue(find_mroute(self, "224.0.0.1", "0.0.0.0", 32,
- table_id=1))
- self.assertFalse(find_mroute(self, "224.0.0.22", "0.0.0.0", 32,
- table_id=1))
+ self.assertTrue(find_mroute(self, "224.0.0.1", "0.0.0.0", 32, table_id=1))
+ self.assertFalse(find_mroute(self, "224.0.0.22", "0.0.0.0", 32, table_id=1))
def verify_general_query(self, p):
ip = p[IP]
self.assertEqual(len(ip.options), 1)
self.assertEqual(ip.options[0].option, 20)
self.assertEqual(ip.proto, 2)
- self.assertEqual(IGMPv3.igmpv3types[rx[IGMPv3].type],
- "Version 3 Membership Report")
+ self.assertEqual(
+ IGMPv3.igmpv3types[rx[IGMPv3].type], "Version 3 Membership Report"
+ )
self.assertEqual(rx[IGMPv3mr].numgrp, len(records))
received = rx[IGMPv3mr].records
self.assertEqual(gr.maddr, r.sg.gaddr)
self.assertEqual(len(gr.srcaddrs), len(r.sg.saddrs))
- self.assertEqual(sorted(gr.srcaddrs),
- sorted(r.sg.saddrs))
+ self.assertEqual(sorted(gr.srcaddrs), sorted(r.sg.saddrs))
def add_group(self, itf, sg, n_pkts=2):
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- hs = VppHostState(self,
- IGMP_FILTER.INCLUDE,
- itf.sw_if_index,
- sg)
+ hs = VppHostState(self, IGMP_FILTER.INCLUDE, itf.sw_if_index, sg)
hs.add_vpp_config()
capture = itf.get_capture(n_pkts, timeout=10)
# reports are transmitted twice due to default rebostness value=2
- self.verify_report(capture[0],
- [IgmpRecord(sg, "Allow New Sources")]),
- self.verify_report(capture[1],
- [IgmpRecord(sg, "Allow New Sources")]),
+ self.verify_report(capture[0], [IgmpRecord(sg, "Allow New Sources")]),
+ self.verify_report(capture[1], [IgmpRecord(sg, "Allow New Sources")]),
return hs
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(hs.sg, "Block Old Sources")])
+ self.verify_report(capture[0], [IgmpRecord(hs.sg, "Block Old Sources")])
def test_igmp_host(self):
- """ IGMP Host functions """
+ """IGMP Host functions"""
#
# Enable interface for host functions
#
- self.vapi.igmp_enable_disable(self.pg0.sw_if_index,
- 1,
- IGMP_MODE.HOST)
+ self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 1, IGMP_MODE.HOST)
#
# Add one S,G of state and expect a state-change event report
# search for the corresponding state created in VPP
dump = self.vapi.igmp_dump(self.pg0.sw_if_index)
self.assertEqual(len(dump), 1)
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.1", "1.1.1.1"))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.1", "1.1.1.1"))
#
# Send a general query (to the all router's address)
# Pad the query with 0 - some devices in the big wild
# internet are prone to this.
#
- p_g = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='224.0.0.1', tos=0xc0) /
- IGMPv3(type="Membership Query", mrcode=100) /
- IGMPv3mq(gaddr="0.0.0.0") /
- Raw(b'\x00' * 10))
+ p_g = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(src=self.pg0.remote_ip4, dst="224.0.0.1", tos=0xC0)
+ / IGMPv3(type="Membership Query", mrcode=100)
+ / IGMPv3mq(gaddr="0.0.0.0")
+ / Raw(b"\x00" * 10)
+ )
self.send(self.pg0, p_g)
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h1.sg, "Mode Is Include")])
+ self.verify_report(capture[0], [IgmpRecord(h1.sg, "Mode Is Include")])
#
# Group specific query
#
- p_gs = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='239.1.1.1', tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Membership Query", mrcode=100) /
- IGMPv3mq(gaddr="239.1.1.1"))
+ p_gs = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="239.1.1.1",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Membership Query", mrcode=100)
+ / IGMPv3mq(gaddr="239.1.1.1")
+ )
self.send(self.pg0, p_gs)
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h1.sg, "Mode Is Include")])
+ self.verify_report(capture[0], [IgmpRecord(h1.sg, "Mode Is Include")])
#
# A group and source specific query, with the source matching
# the source VPP has
#
- p_gs1 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='239.1.1.1', tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Membership Query", mrcode=100) /
- IGMPv3mq(gaddr="239.1.1.1", srcaddrs=["1.1.1.1"]))
+ p_gs1 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="239.1.1.1",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Membership Query", mrcode=100)
+ / IGMPv3mq(gaddr="239.1.1.1", srcaddrs=["1.1.1.1"])
+ )
self.send(self.pg0, p_gs1)
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h1.sg, "Mode Is Include")])
+ self.verify_report(capture[0], [IgmpRecord(h1.sg, "Mode Is Include")])
#
# A group and source specific query that reports more sources
# than the packet actually has.
#
- p_gs2 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='239.1.1.1', tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Membership Query", mrcode=100) /
- IGMPv3mq(gaddr="239.1.1.1", numsrc=4, srcaddrs=["1.1.1.1"]))
+ p_gs2 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="239.1.1.1",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Membership Query", mrcode=100)
+ / IGMPv3mq(gaddr="239.1.1.1", numsrc=4, srcaddrs=["1.1.1.1"])
+ )
self.send_and_assert_no_replies(self.pg0, p_gs2, timeout=10)
# A group and source specific query, with the source NOT matching
# the source VPP has. There should be no response.
#
- p_gs2 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='239.1.1.1', tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Membership Query", mrcode=100) /
- IGMPv3mq(gaddr="239.1.1.1", srcaddrs=["1.1.1.2"]))
+ p_gs2 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="239.1.1.1",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Membership Query", mrcode=100)
+ / IGMPv3mq(gaddr="239.1.1.1", srcaddrs=["1.1.1.2"])
+ )
self.send_and_assert_no_replies(self.pg0, p_gs2, timeout=10)
# one of which matches the source VPP has.
# The report should contain only the source VPP has.
#
- p_gs3 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='239.1.1.1', tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Membership Query", mrcode=100) /
- IGMPv3mq(gaddr="239.1.1.1",
- srcaddrs=["1.1.1.1", "1.1.1.2", "1.1.1.3"]))
+ p_gs3 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="239.1.1.1",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Membership Query", mrcode=100)
+ / IGMPv3mq(gaddr="239.1.1.1", srcaddrs=["1.1.1.1", "1.1.1.2", "1.1.1.3"])
+ )
self.send(self.pg0, p_gs3)
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h1.sg, "Mode Is Include")])
+ self.verify_report(capture[0], [IgmpRecord(h1.sg, "Mode Is Include")])
#
# Two source and group specific queries in quick succession, the
#
self.send(self.pg0, [p_gs2, p_gs1])
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h1.sg, "Mode Is Include")])
+ self.verify_report(capture[0], [IgmpRecord(h1.sg, "Mode Is Include")])
self.send(self.pg0, [p_gs1, p_gs2])
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h1.sg, "Mode Is Include")])
+ self.verify_report(capture[0], [IgmpRecord(h1.sg, "Mode Is Include")])
#
# remove state, expect the report for the removal
#
# A group with multiple sources
#
- h2 = self.add_group(self.pg0,
- IgmpSG("239.1.1.1",
- ["1.1.1.1", "1.1.1.2", "1.1.1.3"]))
+ h2 = self.add_group(
+ self.pg0, IgmpSG("239.1.1.1", ["1.1.1.1", "1.1.1.2", "1.1.1.3"])
+ )
# search for the corresponding state created in VPP
dump = self.vapi.igmp_dump(self.pg0.sw_if_index)
self.assertEqual(len(dump), 3)
for s in h2.sg.saddrs:
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.1", s))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.1", s))
#
# Send a general query (to the all router's address)
# expect VPP to respond with a membership report will all sources
self.send(self.pg0, p_g)
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h2.sg, "Mode Is Include")])
+ self.verify_report(capture[0], [IgmpRecord(h2.sg, "Mode Is Include")])
#
# Group and source specific query; some present some not
#
- p_gs = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst='239.1.1.1', tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Membership Query", mrcode=100) /
- IGMPv3mq(gaddr="239.1.1.1",
- srcaddrs=["1.1.1.1", "1.1.1.2", "1.1.1.4"]))
+ p_gs = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="239.1.1.1",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Membership Query", mrcode=100)
+ / IGMPv3mq(gaddr="239.1.1.1", srcaddrs=["1.1.1.1", "1.1.1.2", "1.1.1.4"])
+ )
self.send(self.pg0, p_gs)
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(
- IgmpSG('239.1.1.1', ["1.1.1.1", "1.1.1.2"]),
- "Mode Is Include")])
+ self.verify_report(
+ capture[0],
+ [
+ IgmpRecord(
+ IgmpSG("239.1.1.1", ["1.1.1.1", "1.1.1.2"]), "Mode Is Include"
+ )
+ ],
+ )
#
# add loads more groups
#
- h3 = self.add_group(self.pg0,
- IgmpSG("239.1.1.2",
- ["2.1.1.1", "2.1.1.2", "2.1.1.3"]))
- h4 = self.add_group(self.pg0,
- IgmpSG("239.1.1.3",
- ["3.1.1.1", "3.1.1.2", "3.1.1.3"]))
- h5 = self.add_group(self.pg0,
- IgmpSG("239.1.1.4",
- ["4.1.1.1", "4.1.1.2", "4.1.1.3"]))
- h6 = self.add_group(self.pg0,
- IgmpSG("239.1.1.5",
- ["5.1.1.1", "5.1.1.2", "5.1.1.3"]))
- h7 = self.add_group(self.pg0,
- IgmpSG("239.1.1.6",
- ["6.1.1.1", "6.1.1.2",
- "6.1.1.3", "6.1.1.4",
- "6.1.1.5", "6.1.1.6",
- "6.1.1.7", "6.1.1.8",
- "6.1.1.9", "6.1.1.10",
- "6.1.1.11", "6.1.1.12",
- "6.1.1.13", "6.1.1.14",
- "6.1.1.15", "6.1.1.16"]))
+ h3 = self.add_group(
+ self.pg0, IgmpSG("239.1.1.2", ["2.1.1.1", "2.1.1.2", "2.1.1.3"])
+ )
+ h4 = self.add_group(
+ self.pg0, IgmpSG("239.1.1.3", ["3.1.1.1", "3.1.1.2", "3.1.1.3"])
+ )
+ h5 = self.add_group(
+ self.pg0, IgmpSG("239.1.1.4", ["4.1.1.1", "4.1.1.2", "4.1.1.3"])
+ )
+ h6 = self.add_group(
+ self.pg0, IgmpSG("239.1.1.5", ["5.1.1.1", "5.1.1.2", "5.1.1.3"])
+ )
+ h7 = self.add_group(
+ self.pg0,
+ IgmpSG(
+ "239.1.1.6",
+ [
+ "6.1.1.1",
+ "6.1.1.2",
+ "6.1.1.3",
+ "6.1.1.4",
+ "6.1.1.5",
+ "6.1.1.6",
+ "6.1.1.7",
+ "6.1.1.8",
+ "6.1.1.9",
+ "6.1.1.10",
+ "6.1.1.11",
+ "6.1.1.12",
+ "6.1.1.13",
+ "6.1.1.14",
+ "6.1.1.15",
+ "6.1.1.16",
+ ],
+ ),
+ )
#
# general query.
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h3.sg, "Mode Is Include"),
- IgmpRecord(h2.sg, "Mode Is Include"),
- IgmpRecord(h6.sg, "Mode Is Include"),
- IgmpRecord(h4.sg, "Mode Is Include"),
- IgmpRecord(h5.sg, "Mode Is Include"),
- IgmpRecord(h7.sg, "Mode Is Include")])
+ self.verify_report(
+ capture[0],
+ [
+ IgmpRecord(h3.sg, "Mode Is Include"),
+ IgmpRecord(h2.sg, "Mode Is Include"),
+ IgmpRecord(h6.sg, "Mode Is Include"),
+ IgmpRecord(h4.sg, "Mode Is Include"),
+ IgmpRecord(h5.sg, "Mode Is Include"),
+ IgmpRecord(h7.sg, "Mode Is Include"),
+ ],
+ )
#
# modify a group to add and remove some sources
#
- h7.sg = IgmpSG("239.1.1.6",
- ["6.1.1.1", "6.1.1.2",
- "6.1.1.5", "6.1.1.6",
- "6.1.1.7", "6.1.1.8",
- "6.1.1.9", "6.1.1.10",
- "6.1.1.11", "6.1.1.12",
- "6.1.1.13", "6.1.1.14",
- "6.1.1.15", "6.1.1.16",
- "6.1.1.17", "6.1.1.18"])
+ h7.sg = IgmpSG(
+ "239.1.1.6",
+ [
+ "6.1.1.1",
+ "6.1.1.2",
+ "6.1.1.5",
+ "6.1.1.6",
+ "6.1.1.7",
+ "6.1.1.8",
+ "6.1.1.9",
+ "6.1.1.10",
+ "6.1.1.11",
+ "6.1.1.12",
+ "6.1.1.13",
+ "6.1.1.14",
+ "6.1.1.15",
+ "6.1.1.16",
+ "6.1.1.17",
+ "6.1.1.18",
+ ],
+ )
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
h7.add_vpp_config()
capture = self.pg0.get_capture(1, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(IgmpSG("239.1.1.6",
- ["6.1.1.17", "6.1.1.18"]),
- "Allow New Sources"),
- IgmpRecord(IgmpSG("239.1.1.6",
- ["6.1.1.3", "6.1.1.4"]),
- "Block Old Sources")])
+ self.verify_report(
+ capture[0],
+ [
+ IgmpRecord(
+ IgmpSG("239.1.1.6", ["6.1.1.17", "6.1.1.18"]), "Allow New Sources"
+ ),
+ IgmpRecord(
+ IgmpSG("239.1.1.6", ["6.1.1.3", "6.1.1.4"]), "Block Old Sources"
+ ),
+ ],
+ )
#
# add an additional groups with many sources so that each group
for i in range(128):
src_list.append("10.1.1.%d" % i)
- h8 = self.add_group(self.pg0,
- IgmpSG("238.1.1.1", src_list))
- h9 = self.add_group(self.pg0,
- IgmpSG("238.1.1.2", src_list))
+ h8 = self.add_group(self.pg0, IgmpSG("238.1.1.1", src_list))
+ h9 = self.add_group(self.pg0, IgmpSG("238.1.1.2", src_list))
self.send(self.pg0, p_g)
capture = self.pg0.get_capture(4, timeout=10)
- self.verify_report(capture[0],
- [IgmpRecord(h3.sg, "Mode Is Include"),
- IgmpRecord(h2.sg, "Mode Is Include"),
- IgmpRecord(h6.sg, "Mode Is Include"),
- IgmpRecord(h4.sg, "Mode Is Include"),
- IgmpRecord(h5.sg, "Mode Is Include")])
- self.verify_report(capture[1],
- [IgmpRecord(h8.sg, "Mode Is Include")])
- self.verify_report(capture[2],
- [IgmpRecord(h7.sg, "Mode Is Include")])
- self.verify_report(capture[3],
- [IgmpRecord(h9.sg, "Mode Is Include")])
+ self.verify_report(
+ capture[0],
+ [
+ IgmpRecord(h3.sg, "Mode Is Include"),
+ IgmpRecord(h2.sg, "Mode Is Include"),
+ IgmpRecord(h6.sg, "Mode Is Include"),
+ IgmpRecord(h4.sg, "Mode Is Include"),
+ IgmpRecord(h5.sg, "Mode Is Include"),
+ ],
+ )
+ self.verify_report(capture[1], [IgmpRecord(h8.sg, "Mode Is Include")])
+ self.verify_report(capture[2], [IgmpRecord(h7.sg, "Mode Is Include")])
+ self.verify_report(capture[3], [IgmpRecord(h9.sg, "Mode Is Include")])
#
# drop the MTU further (so a 128 sized group won't fit)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- h10 = VppHostState(self,
- IGMP_FILTER.INCLUDE,
- self.pg0.sw_if_index,
- IgmpSG("238.1.1.3", src_list))
+ h10 = VppHostState(
+ self,
+ IGMP_FILTER.INCLUDE,
+ self.pg0.sw_if_index,
+ IgmpSG("238.1.1.3", src_list),
+ )
h10.add_vpp_config()
capture = self.pg0.get_capture(2, timeout=10)
# ADD STATE ON MORE INTERFACES
#
- self.vapi.igmp_enable_disable(self.pg0.sw_if_index,
- 0,
- IGMP_MODE.HOST)
+ self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 0, IGMP_MODE.HOST)
def test_igmp_router(self):
- """ IGMP Router Functions """
+ """IGMP Router Functions"""
#
# Drop reports when not enabled
#
- p_j = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0, ttl=1,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Allow New Sources",
- maddr="239.1.1.1", srcaddrs=["10.1.1.1", "10.1.1.2"]))
- p_l = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Block Old Sources",
- maddr="239.1.1.1", srcaddrs=["10.1.1.1", "10.1.1.2"]))
+ p_j = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ ttl=1,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(
+ rtype="Allow New Sources",
+ maddr="239.1.1.1",
+ srcaddrs=["10.1.1.1", "10.1.1.2"],
+ )
+ )
+ p_l = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(
+ rtype="Block Old Sources",
+ maddr="239.1.1.1",
+ srcaddrs=["10.1.1.1", "10.1.1.2"],
+ )
+ )
self.send(self.pg0, p_j)
self.assertFalse(self.vapi.igmp_dump())
#
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- self.vapi.igmp_enable_disable(self.pg0.sw_if_index,
- 1,
- IGMP_MODE.ROUTER)
+ self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 1, IGMP_MODE.ROUTER)
self.vapi.want_igmp_events(1)
#
#
self.send(self.pg0, p_j)
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.1", 1))
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.2", 1))
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.1", 1)
+ )
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.2", 1)
+ )
dump = self.vapi.igmp_dump(self.pg0.sw_if_index)
self.assertEqual(len(dump), 2)
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.1", "10.1.1.1"))
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.1", "10.1.1.2"))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.1", "10.1.1.1"))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.1", "10.1.1.2"))
#
# wait for the per-source timer to expire
# the state should be reaped
# VPP sends a notification that the group has been left
#
- self.assertTrue(wait_for_igmp_event(self, 4, self.pg0,
- "239.1.1.1", "10.1.1.1", 0))
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.2", 0))
+ self.assertTrue(
+ wait_for_igmp_event(self, 4, self.pg0, "239.1.1.1", "10.1.1.1", 0)
+ )
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.2", 0)
+ )
self.assertFalse(self.vapi.igmp_dump())
#
# expired in 3 seconds.
#
self.send(self.pg0, p_j)
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.1", 1))
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.2", 1))
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.1", 1)
+ )
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.2", 1)
+ )
dump = self.vapi.igmp_dump(self.pg0.sw_if_index)
self.assertEqual(len(dump), 2)
self.verify_general_query(capture[0])
self.verify_general_query(capture[1])
- p_cs = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Mode Is Include",
- maddr="239.1.1.1", srcaddrs=["10.1.1.1", "10.1.1.2"]))
+ p_cs = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(
+ rtype="Mode Is Include",
+ maddr="239.1.1.1",
+ srcaddrs=["10.1.1.1", "10.1.1.2"],
+ )
+ )
self.send(self.pg0, p_cs)
self.sleep(2)
dump = self.vapi.igmp_dump(self.pg0.sw_if_index)
self.assertEqual(len(dump), 2)
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.1", "10.1.1.1"))
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.1", "10.1.1.2"))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.1", "10.1.1.1"))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.1", "10.1.1.2"))
#
# wait for the per-source timer to expire
# the state should be reaped
#
- self.assertTrue(wait_for_igmp_event(self, 4, self.pg0,
- "239.1.1.1", "10.1.1.1", 0))
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.2", 0))
+ self.assertTrue(
+ wait_for_igmp_event(self, 4, self.pg0, "239.1.1.1", "10.1.1.1", 0)
+ )
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.2", 0)
+ )
self.assertFalse(self.vapi.igmp_dump())
#
#
self.send(self.pg0, p_j)
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.1", 1))
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.2", 1))
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.1", 1)
+ )
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.2", 1)
+ )
dump = self.vapi.igmp_dump(self.pg0.sw_if_index)
self.assertEqual(len(dump), 2)
self.send(self.pg0, p_l)
capture = self.pg0.get_capture(1, timeout=3)
- self.verify_group_query(capture[0], "239.1.1.1",
- ["10.1.1.1", "10.1.1.2"])
+ self.verify_group_query(capture[0], "239.1.1.1", ["10.1.1.1", "10.1.1.2"])
#
# the group specific query drops the timeout to leave (=1) seconds
#
- self.assertTrue(wait_for_igmp_event(self, 2, self.pg0,
- "239.1.1.1", "10.1.1.1", 0))
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.1", "10.1.1.2", 0))
+ self.assertTrue(
+ wait_for_igmp_event(self, 2, self.pg0, "239.1.1.1", "10.1.1.1", 0)
+ )
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.1", "10.1.1.2", 0)
+ )
self.assertFalse(self.vapi.igmp_dump())
self.assertFalse(self.vapi.igmp_dump())
#
# a TO_EX({}) / IN_EX({}) is treated like a (*,G) join
#
- p_j = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0, ttl=1,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Change To Exclude Mode", maddr="239.1.1.2"))
+ p_j = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ ttl=1,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(rtype="Change To Exclude Mode", maddr="239.1.1.2")
+ )
self.send(self.pg0, p_j)
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.2", "0.0.0.0", 1))
-
- p_j = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0, ttl=1,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Mode Is Exclude", maddr="239.1.1.3"))
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.2", "0.0.0.0", 1)
+ )
+
+ p_j = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ ttl=1,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(rtype="Mode Is Exclude", maddr="239.1.1.3")
+ )
self.send(self.pg0, p_j)
- self.assertTrue(wait_for_igmp_event(self, 1, self.pg0,
- "239.1.1.3", "0.0.0.0", 1))
+ self.assertTrue(
+ wait_for_igmp_event(self, 1, self.pg0, "239.1.1.3", "0.0.0.0", 1)
+ )
#
# A 'allow sources' for {} should be ignored as it should
# never be sent.
#
- p_j = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0, ttl=1,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Allow New Sources", maddr="239.1.1.4"))
+ p_j = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ ttl=1,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(rtype="Allow New Sources", maddr="239.1.1.4")
+ )
self.send(self.pg0, p_j)
dump = self.vapi.igmp_dump(self.pg0.sw_if_index)
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.2", "0.0.0.0"))
- self.assertTrue(find_igmp_state(dump, self.pg0,
- "239.1.1.3", "0.0.0.0"))
- self.assertFalse(find_igmp_state(dump, self.pg0,
- "239.1.1.4", "0.0.0.0"))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.2", "0.0.0.0"))
+ self.assertTrue(find_igmp_state(dump, self.pg0, "239.1.1.3", "0.0.0.0"))
+ self.assertFalse(find_igmp_state(dump, self.pg0, "239.1.1.4", "0.0.0.0"))
#
# a TO_IN({}) and IS_IN({}) are treated like a (*,G) leave
#
self.vapi.cli("set logging class igmp level debug")
- p_l = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0, ttl=1,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Change To Include Mode", maddr="239.1.1.2"))
+ p_l = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ ttl=1,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(rtype="Change To Include Mode", maddr="239.1.1.2")
+ )
self.send(self.pg0, p_l)
- self.assertTrue(wait_for_igmp_event(self, 2, self.pg0,
- "239.1.1.2", "0.0.0.0", 0))
-
- p_l = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst="224.0.0.22", tos=0xc0, ttl=1,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype="Mode Is Include", maddr="239.1.1.3"))
+ self.assertTrue(
+ wait_for_igmp_event(self, 2, self.pg0, "239.1.1.2", "0.0.0.0", 0)
+ )
+
+ p_l = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / IP(
+ src=self.pg0.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ ttl=1,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(rtype="Mode Is Include", maddr="239.1.1.3")
+ )
self.send(self.pg0, p_l)
- self.assertTrue(wait_for_igmp_event(self, 2, self.pg0,
- "239.1.1.3", "0.0.0.0", 0))
+ self.assertTrue(
+ wait_for_igmp_event(self, 2, self.pg0, "239.1.1.3", "0.0.0.0", 0)
+ )
self.assertFalse(self.vapi.igmp_dump(self.pg0.sw_if_index))
#
# disable router config
#
- self.vapi.igmp_enable_disable(self.pg0.sw_if_index,
- 0,
- IGMP_MODE.ROUTER)
+ self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 0, IGMP_MODE.ROUTER)
def _create_igmpv3_pck(self, itf, rtype, maddr, srcaddrs):
- p = (Ether(dst=itf.local_mac, src=itf.remote_mac) /
- IP(src=itf.remote_ip4, dst="224.0.0.22", tos=0xc0, ttl=1,
- options=[IPOption(copy_flag=1, optclass="control",
- option="router_alert")]) /
- IGMPv3(type="Version 3 Membership Report") /
- IGMPv3mr(numgrp=1) /
- IGMPv3gr(rtype=rtype,
- maddr=maddr, srcaddrs=srcaddrs))
+ p = (
+ Ether(dst=itf.local_mac, src=itf.remote_mac)
+ / IP(
+ src=itf.remote_ip4,
+ dst="224.0.0.22",
+ tos=0xC0,
+ ttl=1,
+ options=[
+ IPOption(copy_flag=1, optclass="control", option="router_alert")
+ ],
+ )
+ / IGMPv3(type="Version 3 Membership Report")
+ / IGMPv3mr(numgrp=1)
+ / IGMPv3gr(rtype=rtype, maddr=maddr, srcaddrs=srcaddrs)
+ )
return p
def test_igmp_proxy_device(self):
- """ IGMP proxy device """
+ """IGMP proxy device"""
self.pg2.admin_down()
self.pg2.unconfig_ip4()
self.pg2.set_table_ip4(0)
self.pg2.config_ip4()
self.pg2.admin_up()
- self.vapi.cli('test igmp timers query 10 src 3 leave 1')
+ self.vapi.cli("test igmp timers query 10 src 3 leave 1")
# enable IGMP
self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 1, IGMP_MODE.HOST)
- self.vapi.igmp_enable_disable(self.pg1.sw_if_index, 1,
- IGMP_MODE.ROUTER)
- self.vapi.igmp_enable_disable(self.pg2.sw_if_index, 1,
- IGMP_MODE.ROUTER)
+ self.vapi.igmp_enable_disable(self.pg1.sw_if_index, 1, IGMP_MODE.ROUTER)
+ self.vapi.igmp_enable_disable(self.pg2.sw_if_index, 1, IGMP_MODE.ROUTER)
# create IGMP proxy device
self.vapi.igmp_proxy_device_add_del(0, self.pg0.sw_if_index, 1)
- self.vapi.igmp_proxy_device_add_del_interface(0,
- self.pg1.sw_if_index, 1)
- self.vapi.igmp_proxy_device_add_del_interface(0,
- self.pg2.sw_if_index, 1)
+ self.vapi.igmp_proxy_device_add_del_interface(0, self.pg1.sw_if_index, 1)
+ self.vapi.igmp_proxy_device_add_del_interface(0, self.pg2.sw_if_index, 1)
# send join on pg1. join should be proxied by pg0
- p_j = self._create_igmpv3_pck(self.pg1, "Allow New Sources",
- "239.1.1.1", ["10.1.1.1", "10.1.1.2"])
+ p_j = self._create_igmpv3_pck(
+ self.pg1, "Allow New Sources", "239.1.1.1", ["10.1.1.1", "10.1.1.2"]
+ )
self.send(self.pg1, p_j)
capture = self.pg0.get_capture(1, timeout=1)
- self.verify_report(capture[0], [IgmpRecord(IgmpSG("239.1.1.1",
- ["10.1.1.1", "10.1.1.2"]), "Allow New Sources")])
+ self.verify_report(
+ capture[0],
+ [
+ IgmpRecord(
+ IgmpSG("239.1.1.1", ["10.1.1.1", "10.1.1.2"]), "Allow New Sources"
+ )
+ ],
+ )
self.assertTrue(find_mroute(self, "239.1.1.1", "0.0.0.0", 32))
# send join on pg2. join should be proxied by pg0.
# the group should contain only 10.1.1.3 as
# 10.1.1.1 was already reported
- p_j = self._create_igmpv3_pck(self.pg2, "Allow New Sources",
- "239.1.1.1", ["10.1.1.1", "10.1.1.3"])
+ p_j = self._create_igmpv3_pck(
+ self.pg2, "Allow New Sources", "239.1.1.1", ["10.1.1.1", "10.1.1.3"]
+ )
self.send(self.pg2, p_j)
capture = self.pg0.get_capture(1, timeout=1)
- self.verify_report(capture[0], [IgmpRecord(IgmpSG("239.1.1.1",
- ["10.1.1.3"]), "Allow New Sources")])
+ self.verify_report(
+ capture[0],
+ [IgmpRecord(IgmpSG("239.1.1.1", ["10.1.1.3"]), "Allow New Sources")],
+ )
self.assertTrue(find_mroute(self, "239.1.1.1", "0.0.0.0", 32))
# send leave on pg2. leave for 10.1.1.3 should be proxyed
# as pg2 was the only interface interested in 10.1.1.3
- p_l = self._create_igmpv3_pck(self.pg2, "Block Old Sources",
- "239.1.1.1", ["10.1.1.3"])
+ p_l = self._create_igmpv3_pck(
+ self.pg2, "Block Old Sources", "239.1.1.1", ["10.1.1.3"]
+ )
self.send(self.pg2, p_l)
capture = self.pg0.get_capture(1, timeout=2)
- self.verify_report(capture[0], [IgmpRecord(IgmpSG("239.1.1.1",
- ["10.1.1.3"]), "Block Old Sources")])
+ self.verify_report(
+ capture[0],
+ [IgmpRecord(IgmpSG("239.1.1.1", ["10.1.1.3"]), "Block Old Sources")],
+ )
self.assertTrue(find_mroute(self, "239.1.1.1", "0.0.0.0", 32))
# disable igmp on pg1 (also removes interface from proxy device)
# proxy leave for 10.1.1.2. pg2 is still interested in 10.1.1.1
self.pg_enable_capture(self.pg_interfaces)
- self.vapi.igmp_enable_disable(self.pg1.sw_if_index, 0,
- IGMP_MODE.ROUTER)
+ self.vapi.igmp_enable_disable(self.pg1.sw_if_index, 0, IGMP_MODE.ROUTER)
capture = self.pg0.get_capture(1, timeout=1)
- self.verify_report(capture[0], [IgmpRecord(IgmpSG("239.1.1.1",
- ["10.1.1.2"]), "Block Old Sources")])
+ self.verify_report(
+ capture[0],
+ [IgmpRecord(IgmpSG("239.1.1.1", ["10.1.1.2"]), "Block Old Sources")],
+ )
self.assertTrue(find_mroute(self, "239.1.1.1", "0.0.0.0", 32))
# disable IGMP on pg0 and pg1.
# disabling IGMP on pg0 (proxy device upstream interface)
# removes this proxy device
self.vapi.igmp_enable_disable(self.pg0.sw_if_index, 0, IGMP_MODE.HOST)
- self.vapi.igmp_enable_disable(self.pg2.sw_if_index, 0,
- IGMP_MODE.ROUTER)
+ self.vapi.igmp_enable_disable(self.pg2.sw_if_index, 0, IGMP_MODE.ROUTER)
self.assertFalse(find_mroute(self, "239.1.1.1", "0.0.0.0", 32))
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff