class TestL2fib(VppTestCase):
- """ L2 FIB Test Case """
+ """L2 FIB Test Case"""
@classmethod
def bd_ifs(cls, bd_id):
- return range((bd_id - 1) * cls.n_ifs_per_bd,
- bd_id * cls.n_ifs_per_bd - 1)
+ return range((bd_id - 1) * cls.n_ifs_per_bd, bd_id * cls.n_ifs_per_bd - 1)
@classmethod
def setUpClass(cls):
ifs = cls.bd_ifs(bd_id)
for j in ifs:
cls.flows[cls.pg_interfaces[j]] = [
- cls.pg_interfaces[x] for x in ifs if x != j]
+ cls.pg_interfaces[x] for x in ifs if x != j
+ ]
# Packet sizes
cls.pg_if_packet_sizes = [64, 512, 1518, 9018]
for bd_id in n_brs:
# Create BD with MAC learning and unknown unicast flooding
# disabled and put interfaces to this BD
- cls.vapi.bridge_domain_add_del(bd_id=bd_id, uu_flood=0,
- learn=0)
+ cls.vapi.bridge_domain_add_del(bd_id=bd_id, uu_flood=0, learn=0)
ifs = [cls.pg_interfaces[i] for i in cls.bd_ifs(bd_id)]
for pg_if in ifs:
cls.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=pg_if.sw_if_index, bd_id=bd_id)
+ rx_sw_if_index=pg_if.sw_if_index, bd_id=bd_id
+ )
# Set up all interfaces
for i in cls.pg_interfaces:
super(TestL2fib, self).tearDown()
if not self.vpp_dead:
for bd_id in self.n_brs:
- self.logger.info(self.vapi.ppcli("show bridge-domain %s detail"
- % bd_id))
+ self.logger.info(
+ self.vapi.ppcli("show bridge-domain %s detail" % bd_id)
+ )
def show_commands_at_teardown(self):
self.logger.info(self.vapi.ppcli("show l2fib verbose"))
for pg_if in self.pg_interfaces:
swif = pg_if.sw_if_index
- def mac(j): return "00:00:%02x:ff:%02x:%02x" % (subnet, swif, j)
+ def mac(j):
+ return "00:00:%02x:ff:%02x:%02x" % (subnet, swif, j)
- def ip(j): return "172.%02u.1%02x.%u" % (subnet, swif, j)
+ def ip(j):
+ return "172.%02u.1%02x.%u" % (subnet, swif, j)
+
+ def h(j):
+ return Host(mac(j), ip(j))
- def h(j): return Host(mac(j), ip(j))
hosts[swif] = [h(j) for j in range(n_hosts_per_if)]
return hosts
ifs = [self.pg_interfaces[i] for i in self.bd_ifs(bd_id)]
for pg_if in ifs:
swif = pg_if.sw_if_index
- packets = [Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac)
- for host in hosts[swif]]
+ packets = [
+ Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac) for host in hosts[swif]
+ ]
pg_if.add_stream(packets)
self.logger.info("Sending broadcast eth frames for MAC learning")
self.pg_start()
for pg_if in ifs:
swif = pg_if.sw_if_index
for host in hosts[swif]:
- self.vapi.l2fib_add_del(
- mac_pton(host.mac), bd_id, swif, static_mac=1)
+ self.vapi.l2fib_add_del(mac_pton(host.mac), bd_id, swif, static_mac=1)
def delete_l2_fib_entry(self, bd_id, hosts):
"""
for pg_if in ifs:
swif = pg_if.sw_if_index
for host in hosts[swif]:
- self.vapi.l2fib_add_del(
- mac_pton(host.mac), bd_id, swif, is_add=0)
+ self.vapi.l2fib_add_del(mac_pton(host.mac), bd_id, swif, is_add=0)
def flush_int(self, swif, learned_hosts):
"""
src_host = random.choice(src_hosts)
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)
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)
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, bd_id, src_hosts, dst_hosts):
# Test
ifs = [self.pg_interfaces[i] for i in self.bd_ifs(bd_id)]
for i in ifs:
pkts = self.create_stream(
- i, self.pg_if_packet_sizes,
+ i,
+ self.pg_if_packet_sizes,
if_src_hosts=src_hosts,
- if_dst_hosts=dst_hosts)
+ if_dst_hosts=dst_hosts,
+ )
if pkts:
i.add_stream(pkts)
ifs = [self.pg_interfaces[i] for i in self.bd_ifs(bd_id)]
for i in ifs:
pkts = self.create_stream(
- i, self.pg_if_packet_sizes,
+ i,
+ self.pg_if_packet_sizes,
if_src_hosts=src_hosts,
- if_dst_hosts=dst_hosts)
+ if_dst_hosts=dst_hosts,
+ )
if pkts:
i.add_stream(pkts)
timeout = 0.1
def test_l2_fib_program100(self):
- """ L2 FIB - program 100 MACs
- """
+ """L2 FIB - program 100 MACs"""
bd_id = 1
hosts = self.create_hosts(100, subnet=17)
self.config_l2_fib_entries(bd_id, hosts)
self.run_verify_test(bd_id, hosts, hosts)
def test_l2_fib_program100_delete12(self):
- """ L2 FIB - program 100, delete 12 MACs
- """
+ """L2 FIB - program 100, delete 12 MACs"""
bd_id = 1
hosts = self.create_hosts(100, subnet=17)
self.config_l2_fib_entries(bd_id, hosts)
self.run_verify_negat_test(bd_id, hosts, del_hosts)
def test_l2_fib_program100_add100(self):
- """ L2 FIB - program 100, add 100 MACs
- """
+ """L2 FIB - program 100, add 100 MACs"""
bd_id = 1
hosts = self.create_hosts(100, subnet=17)
self.config_l2_fib_entries(bd_id, hosts)
self.run_verify_test(bd_id, hosts, hosts2)
def test_l2_fib_program10_learn10(self):
- """ L2 FIB - program 10 MACs, learn 10
- """
+ """L2 FIB - program 10 MACs, learn 10"""
hosts = self.create_hosts(20, subnet=35)
lhosts = self.split_hosts(hosts, 10)
self.run_verify_test(bd2, lhosts, hosts)
def test_l2_fib_flush_int(self):
- """ L2 FIB - flush interface
- """
+ """L2 FIB - flush interface"""
hosts = self.create_hosts(20, subnet=36)
lhosts = self.split_hosts(hosts, 10)
self.run_verify_negat_test(bd1, hosts, flushed)
def test_l2_fib_flush_bd(self):
- """ L2 FIB - flush BD
- """
+ """L2 FIB - flush BD"""
hosts = self.create_hosts(20, subnet=37)
lhosts = self.split_hosts(hosts, 10)
self.run_verify_negat_test(bd1, hosts, flushed)
def test_l2_fib_flush_all(self):
- """ L2 FIB - flush all
- """
+ """L2 FIB - flush all"""
hosts = self.create_hosts(20, subnet=38)
lhosts = self.split_hosts(hosts, 10)
self.run_verify_negat_test(bd2, hosts, lhosts)
def test_l2_fib_mac_learn_evs(self):
- """ L2 FIB - mac learning events
- """
+ """L2 FIB - mac learning events"""
bd1 = 1
hosts = self.create_hosts(10, subnet=39)
evs = self.vapi.collect_events()
action = VppEnum.vl_api_mac_event_action_t.MAC_EVENT_ACTION_API_ADD
learned_macs = {
- e.mac[i].mac_addr.packed for e in evs for i in range(e.n_macs)
- if e.mac[i].action == action}
- macs = {h.bin_mac for swif in self.bd_ifs(bd1)
- for h in hosts[self.pg_interfaces[swif].sw_if_index]}
+ e.mac[i].mac_addr.packed
+ for e in evs
+ for i in range(e.n_macs)
+ if e.mac[i].action == action
+ }
+ macs = {
+ h.bin_mac
+ for swif in self.bd_ifs(bd1)
+ for h in hosts[self.pg_interfaces[swif].sw_if_index]
+ }
self.vapi.want_l2_macs_events(enable_disable=0)
self.assertEqual(len(learned_macs ^ macs), 0)
def test_l2_fib_mac_learn_evs2(self):
- """ L2 FIB - mac learning events using want_l2_macs_events2
- """
+ """L2 FIB - mac learning events using want_l2_macs_events2"""
bd1 = 1
hosts = self.create_hosts(10, subnet=39)
evs = self.vapi.collect_events()
action = VppEnum.vl_api_mac_event_action_t.MAC_EVENT_ACTION_API_ADD
learned_macs = {
- e.mac[i].mac_addr.packed for e in evs for i in range(e.n_macs)
- if e.mac[i].action == action}
- macs = {h.bin_mac for swif in self.bd_ifs(bd1)
- for h in hosts[self.pg_interfaces[swif].sw_if_index]}
+ e.mac[i].mac_addr.packed
+ for e in evs
+ for i in range(e.n_macs)
+ if e.mac[i].action == action
+ }
+ macs = {
+ h.bin_mac
+ for swif in self.bd_ifs(bd1)
+ for h in hosts[self.pg_interfaces[swif].sw_if_index]
+ }
self.vapi.want_l2_macs_events2(enable_disable=0)
self.assertEqual(len(learned_macs ^ macs), 0)
def test_l2_fib_macs_learn_max(self):
- """ L2 FIB - mac learning max macs in event
- """
+ """L2 FIB - mac learning max macs in event"""
bd1 = 1
hosts = self.create_hosts(10, subnet=40)
self.assertGreater(len(evs), 0)
action = VppEnum.vl_api_mac_event_action_t.MAC_EVENT_ACTION_API_ADD
learned_macs = {
- e.mac[i].mac_addr.packed for e in evs for i in range(e.n_macs)
- if e.mac[i].action == action}
- macs = {h.bin_mac for swif in self.bd_ifs(bd1)
- for h in hosts[self.pg_interfaces[swif].sw_if_index]}
+ e.mac[i].mac_addr.packed
+ for e in evs
+ for i in range(e.n_macs)
+ if e.mac[i].action == action
+ }
+ macs = {
+ h.bin_mac
+ for swif in self.bd_ifs(bd1)
+ for h in hosts[self.pg_interfaces[swif].sw_if_index]
+ }
for e in evs:
self.assertLess(len(e), ev_macs * 10)
self.assertEqual(len(learned_macs ^ macs), 0)
def test_l2_fib_macs_learn_max2(self):
- """ L2 FIB - mac learning max macs in event using want_l2_macs_events2
- """
+ """L2 FIB - mac learning max macs in event using want_l2_macs_events2"""
bd1 = 1
hosts = self.create_hosts(10, subnet=40)
self.assertGreater(len(evs), 0)
action = VppEnum.vl_api_mac_event_action_t.MAC_EVENT_ACTION_API_ADD
learned_macs = {
- e.mac[i].mac_addr.packed for e in evs for i in range(e.n_macs)
- if e.mac[i].action == action}
- macs = {h.bin_mac for swif in self.bd_ifs(bd1)
- for h in hosts[self.pg_interfaces[swif].sw_if_index]}
+ e.mac[i].mac_addr.packed
+ for e in evs
+ for i in range(e.n_macs)
+ if e.mac[i].action == action
+ }
+ macs = {
+ h.bin_mac
+ for swif in self.bd_ifs(bd1)
+ for h in hosts[self.pg_interfaces[swif].sw_if_index]
+ }
for e in evs:
self.assertLess(len(e), ev_macs * 10)
self.assertEqual(len(learned_macs ^ macs), 0)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff