6 from framework import VppTestCase, VppTestRunner
7 from vpp_ip_route import VppIpRoute, VppRoutePath
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether
11 from scapy.layers.inet import IP, UDP
14 class TestL2Flood(VppTestCase):
18 super(TestL2Flood, self).setUp()
20 # 12 l2 interface and one l3
21 self.create_pg_interfaces(range(13))
22 self.create_loopback_interfaces(1)
24 for i in self.pg_interfaces:
26 for i in self.lo_interfaces:
29 self.pg12.config_ip4()
30 self.pg12.resolve_arp()
31 self.loop0.config_ip4()
34 self.pg12.unconfig_ip4()
35 self.loop0.unconfig_ip4()
37 for i in self.pg_interfaces:
39 for i in self.lo_interfaces:
41 super(TestL2Flood, self).tearDown()
44 """ L2 Flood Tests """
47 # Create a single bridge Domain
49 self.vapi.bridge_domain_add_del(1)
52 # add each interface to the BD. 3 interfaces per split horizon group
54 for i in self.pg_interfaces[0:4]:
55 self.vapi.sw_interface_set_l2_bridge(i.sw_if_index, 1, 0)
56 for i in self.pg_interfaces[4:8]:
57 self.vapi.sw_interface_set_l2_bridge(i.sw_if_index, 1, 1)
58 for i in self.pg_interfaces[8:12]:
59 self.vapi.sw_interface_set_l2_bridge(i.sw_if_index, 1, 2)
60 for i in self.lo_interfaces:
61 self.vapi.sw_interface_set_l2_bridge(i.sw_if_index, 1, 2, bvi=1)
63 p = (Ether(dst="ff:ff:ff:ff:ff:ff",
64 src="00:00:de:ad:be:ef") /
65 IP(src="10.10.10.10", dst="1.1.1.1") /
66 UDP(sport=1234, dport=1234) /
70 # input on pg0 expect copies on pg1->11
71 # this is in SHG=0 so its flooded to all, expect the pg0 since that's
74 self.pg0.add_stream(p*65)
75 self.pg_enable_capture(self.pg_interfaces)
78 for i in self.pg_interfaces[1:12]:
79 rx0 = i.get_capture(65, timeout=1)
82 # input on pg4 (SHG=1) expect copies on pg0->3 (SHG=0)
85 self.pg4.add_stream(p*65)
86 self.pg_enable_capture(self.pg_interfaces)
89 for i in self.pg_interfaces[:4]:
90 rx0 = i.get_capture(65, timeout=1)
91 for i in self.pg_interfaces[8:12]:
92 rx0 = i.get_capture(65, timeout=1)
93 for i in self.pg_interfaces[4:8]:
94 i.assert_nothing_captured(remark="Different SH group")
97 # An IP route so the packet that hits the BVI is sent out of pg12
99 ip_route = VppIpRoute(self, "1.1.1.1", 32,
100 [VppRoutePath(self.pg12.remote_ip4,
101 self.pg12.sw_if_index)])
102 ip_route.add_vpp_config()
104 self.logger.info(self.vapi.cli("sh bridge 1 detail"))
107 # input on pg0 expect copies on pg1->12
108 # this is in SHG=0 so its flooded to all, expect the pg0 since that's
111 self.pg0.add_stream(p*65)
112 self.pg_enable_capture(self.pg_interfaces)
115 for i in self.pg_interfaces[1:]:
116 rx0 = i.get_capture(65, timeout=1)
119 # input on pg4 (SHG=1) expect copies on pg0->3 (SHG=0)
120 # and pg8->12 (SHG=2)
122 self.pg4.add_stream(p*65)
123 self.pg_enable_capture(self.pg_interfaces)
126 for i in self.pg_interfaces[:4]:
127 rx0 = i.get_capture(65, timeout=1)
128 for i in self.pg_interfaces[8:13]:
129 rx0 = i.get_capture(65, timeout=1)
130 for i in self.pg_interfaces[4:8]:
131 i.assert_nothing_captured(remark="Different SH group")
136 for i in self.pg_interfaces[:12]:
137 self.vapi.sw_interface_set_l2_bridge(i.sw_if_index, 1, enable=0)
138 for i in self.lo_interfaces:
139 self.vapi.sw_interface_set_l2_bridge(i.sw_if_index, 1, 2,
142 self.vapi.bridge_domain_add_del(1, is_add=0)
145 if __name__ == '__main__':
146 unittest.main(testRunner=VppTestRunner)