-#!/usr/bin/env python
+#!/usr/bin/env python3
"""L2 FIB Test Case HLD:
**config 1**
from framework import VppTestCase, VppTestRunner
from util import Host, ppp
+from vpp_papi import mac_pton, VppEnum
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)
+ return range((bd_id - 1) * cls.n_ifs_per_bd, bd_id * cls.n_ifs_per_bd - 1)
@classmethod
def setUpClass(cls):
n_brs = cls.n_brs = range(1, 3)
cls.n_ifs_per_bd = 4
n_ifs = range(cls.n_ifs_per_bd * len(cls.n_brs))
- # Create 4 pg interfaces
+ # Create pg interfaces
cls.create_pg_interfaces(n_ifs)
cls.flows = dict()
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(pg_if.sw_if_index,
- bd_id=bd_id)
+ cls.vapi.sw_interface_set_l2_bridge(
+ rx_sw_if_index=pg_if.sw_if_index, bd_id=bd_id
+ )
# Set up all interfaces
for i in cls.pg_interfaces:
i.admin_up()
-
- # Mapping between packet-generator index and lists of test hosts
- cls.hosts = dict()
- cls.learned_hosts = dict()
- cls.fib_hosts = dict()
- cls.deleted_hosts = dict()
- for pg_if in cls.pg_interfaces:
- swif = pg_if.sw_if_index
- cls.hosts[swif] = []
- cls.learned_hosts[swif] = []
- cls.fib_hosts[swif] = []
- cls.deleted_hosts[swif] = []
-
except Exception:
super(TestL2fib, cls).tearDownClass()
raise
+ @classmethod
+ def tearDownClass(cls):
+ super(TestL2fib, cls).tearDownClass()
+
def setUp(self):
super(TestL2fib, self).setUp()
self.reset_packet_infos()
"""
super(TestL2fib, self).tearDown()
if not self.vpp_dead:
- self.logger.info(self.vapi.ppcli("show l2fib verbose"))
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"))
def create_hosts(self, n_hosts_per_if, subnet):
"""
interfaces. Create host IPv4 address for every host MAC address.
:param int n_hosts_per_if: Number of per interface hosts to
- create MAC/IPv4 addresses for.
+ create MAC/IPv4 addresses for.
"""
+ hosts = dict()
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))
- self.hosts[swif] = [h(j) for j in range(n_hosts_per_if)]
+ hosts[swif] = [h(j) for j in range(n_hosts_per_if)]
+ return hosts
- def learn_hosts(self, bd_id, n_hosts_per_if):
+ def split_hosts(self, hosts, n):
+ splits = dict()
+ for pg_if in self.pg_interfaces:
+ swif = pg_if.sw_if_index
+ splits[swif] = hosts[swif][:n]
+ hosts[swif] = hosts[swif][n:]
+ return splits
+
+ def learn_hosts(self, bd_id, hosts):
"""
- Create L2 MAC packet stream with host MAC addresses per interface to
+ Create and send per interface L2 MAC broadcast packet stream to
let the bridge domain learn these MAC addresses.
:param int bd_id: BD to teach
- :param int n_hosts_per_if: number of hosts
+ :param dict hosts: dict of hosts per interface
"""
- self.vapi.bridge_flags(bd_id, 1, 1)
+ self.vapi.bridge_flags(bd_id=bd_id, is_set=1, flags=1)
ifs = [self.pg_interfaces[i] for i in self.bd_ifs(bd_id)]
for pg_if in ifs:
swif = pg_if.sw_if_index
- hosts = self.hosts[swif]
- lhosts = self.learned_hosts[swif]
- packets = []
- for j in range(n_hosts_per_if):
- host = hosts.pop()
- lhosts.append(host)
- packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
- packets.append(packet)
+ 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()
- def config_l2_fib_entries(self, bd_id, n_hosts_per_if):
+ def config_l2_fib_entries(self, bd_id, hosts):
"""
Config required number of L2 FIB entries.
ifs = [self.pg_interfaces[i] for i in self.bd_ifs(bd_id)]
for pg_if in ifs:
swif = pg_if.sw_if_index
- hosts = self.hosts[swif]
- fhosts = self.fib_hosts[swif]
- for j in range(n_hosts_per_if):
- host = hosts.pop()
- self.vapi.l2fib_add_del(
- host.mac, bd_id, swif, static_mac=1)
- fhosts.append(host)
- # del hosts[0]
- self.logger.info("Configure %d L2 FIB entries .." %
- len(self.pg_interfaces) * n_hosts_per_if)
- self.logger.info(self.vapi.ppcli("show l2fib"))
+ for host in hosts[swif]:
+ self.vapi.l2fib_add_del(mac_pton(host.mac), bd_id, swif, static_mac=1)
- def delete_l2_fib_entry(self, bd_id, n_hosts_per_if):
+ def delete_l2_fib_entry(self, bd_id, hosts):
"""
Delete required number of L2 FIB entries.
ifs = [self.pg_interfaces[i] for i in self.bd_ifs(bd_id)]
for pg_if in ifs:
swif = pg_if.sw_if_index
- hosts = self.fib_hosts[swif]
- dhosts = self.deleted_hosts[swif]
- for j in range(n_hosts_per_if):
- host = hosts.pop()
- self.vapi.l2fib_add_del(
- host.mac, bd_id, swif, is_add=0)
- dhosts.append(host)
- self.logger.info(self.vapi.ppcli("show l2fib"))
+ for host in hosts[swif]:
+ self.vapi.l2fib_add_del(mac_pton(host.mac), bd_id, swif, is_add=0)
- def flush_int(self, swif):
+ def flush_int(self, swif, learned_hosts):
"""
Flush swif L2 FIB entries.
"""
flushed = dict()
self.vapi.l2fib_flush_int(swif)
- self.deleted_hosts[swif] = self.learned_hosts[swif] + \
- self.deleted_hosts[swif]
- flushed[swif] = self.learned_hosts[swif]
- self.learned_hosts[swif] = []
- self.logger.info(self.vapi.ppcli("show l2fib"))
+ flushed[swif] = learned_hosts[swif]
+ learned_hosts[swif] = []
return flushed
- def flush_bd(self, bd_id):
+ def flush_bd(self, bd_id, learned_hosts):
"""
Flush bd_id L2 FIB entries.
ifs = [self.pg_interfaces[i] for i in self.bd_ifs(bd_id)]
for pg_if in ifs:
swif = pg_if.sw_if_index
- self.deleted_hosts[swif] = self.learned_hosts[swif] + \
- self.deleted_hosts[swif]
- flushed[swif] = self.learned_hosts[swif]
- self.learned_hosts[swif] = []
- self.logger.info(self.vapi.ppcli("show l2fib"))
+ flushed[swif] = learned_hosts[swif]
+ learned_hosts[swif] = []
return flushed
def flush_all(self):
Flush All L2 FIB entries.
"""
self.vapi.l2fib_flush_all()
- flushed = dict()
- for pg_if in self.pg_interfaces:
- swif = pg_if.sw_if_index
- self.deleted_hosts[swif] = self.learned_hosts[swif] + \
- self.deleted_hosts[swif]
- flushed[swif] = self.learned_hosts[swif]
- self.learned_hosts[swif] = []
- self.logger.info(self.vapi.ppcli("show l2fib"))
- return flushed
- def create_stream(self, src_if, packet_sizes, if_src_hosts=None,
- if_dst_hosts=None):
+ def create_stream(self, src_if, packet_sizes, if_src_hosts, if_dst_hosts):
"""
Create input packet stream for defined interface using hosts or
deleted_hosts list.
:param boolean deleted: Set to True if deleted_hosts list required.
:return: Stream of packets.
"""
- if not if_src_hosts:
- if_src_hosts = self.fib_hosts
- if not if_dst_hosts:
- if_dst_hosts = self.fib_hosts
src_hosts = if_src_hosts[src_if.sw_if_index]
if not src_hosts:
return []
if dst_swif not in if_dst_hosts:
continue
dst_hosts = if_dst_hosts[dst_swif]
- for i in range(0, len(dst_hosts)):
- dst_host = dst_hosts[i]
+ for dst_host in dst_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)
last_info[i.sw_if_index] = None
dst_sw_if_index = pg_if.sw_if_index
for packet in capture:
- payload_info = self.payload_to_info(str(packet[Raw]))
+ payload_info = self.payload_to_info(packet[Raw])
try:
ip = packet[IP]
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)
self.assertEqual(ip.dst, saved_packet[IP].dst)
self.assertEqual(udp.sport, saved_packet[UDP].sport)
self.assertEqual(udp.dport, saved_packet[UDP].dport)
- except:
+ except BaseException:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
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, dst_hosts=None):
+ def run_verify_test(self, bd_id, src_hosts, dst_hosts):
# Test
# Create incoming packet streams for packet-generator interfaces
- if not dst_hosts:
- dst_hosts = self.fib_hosts
self.reset_packet_infos()
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, if_dst_hosts=dst_hosts)
- i.add_stream(pkts)
+ i,
+ self.pg_if_packet_sizes,
+ if_src_hosts=src_hosts,
+ if_dst_hosts=dst_hosts,
+ )
+ if pkts:
+ i.add_stream(pkts)
- self.vapi.bridge_flags(bd_id, 0, 1)
+ self.vapi.bridge_flags(bd_id=bd_id, is_set=0, flags=1)
# Enable packet capture and start packet sending
self.pg_enable_capture(ifs)
self.pg_start()
self.logger.info("Verifying capture on interface %s" % i.name)
self.verify_capture(i, capture)
- def run_verify_negat_test(self, bd_id, dst_hosts=None):
+ def run_verify_negat_test(self, bd_id, src_hosts, dst_hosts):
# Test
# Create incoming packet streams for packet-generator interfaces for
# deleted MAC addresses
- if not dst_hosts:
- dst_hosts = self.deleted_hosts
self.reset_packet_infos()
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, if_dst_hosts=dst_hosts)
+ i,
+ self.pg_if_packet_sizes,
+ if_src_hosts=src_hosts,
+ if_dst_hosts=dst_hosts,
+ )
if pkts:
i.add_stream(pkts)
- self.vapi.bridge_flags(bd_id, 0, 1)
+ self.vapi.bridge_flags(bd_id=bd_id, is_set=0, flags=1)
# Enable packet capture and start packet sending
self.pg_enable_capture(ifs)
self.pg_start()
i.assert_nothing_captured(remark="outgoing interface")
timeout = 0.1
- def test_l2_fib_01(self):
- """ L2 FIB test 1 - program 100 MAC addresses
- """
- # Config 1
- # Create test host entries
- self.create_hosts(100, subnet=17)
-
- # Add first 100 MAC entries to L2 FIB
- self.config_l2_fib_entries(bd_id=1, n_hosts_per_if=100)
-
- # Test 1
- self.run_verify_test(bd_id=1)
-
- def test_l2_fib_02(self):
- """ L2 FIB test 2 - delete 12 MAC entries
- """
- # Config 2
- # Delete 12 MAC entries per interface from L2 FIB
- self.delete_l2_fib_entry(bd_id=1, n_hosts_per_if=12)
-
- # Test 2a
- self.run_verify_test(bd_id=1)
-
- # Verify 2a
- self.run_verify_negat_test(bd_id=1)
+ def test_l2_fib_program100(self):
+ """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"""
+ bd_id = 1
+ hosts = self.create_hosts(100, subnet=17)
+ self.config_l2_fib_entries(bd_id, hosts)
+ del_hosts = self.split_hosts(hosts, 12)
+ self.delete_l2_fib_entry(bd_id, del_hosts)
+
+ self.run_verify_test(bd_id, hosts, 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"""
+ bd_id = 1
+ hosts = self.create_hosts(100, subnet=17)
+ self.config_l2_fib_entries(bd_id, hosts)
+ hosts2 = self.create_hosts(100, subnet=22)
+ self.config_l2_fib_entries(bd_id, hosts2)
+ self.run_verify_test(bd_id, hosts, hosts2)
+
+ def test_l2_fib_program10_learn10(self):
+ """L2 FIB - program 10 MACs, learn 10"""
+ hosts = self.create_hosts(20, subnet=35)
+ lhosts = self.split_hosts(hosts, 10)
+
+ bd1 = 1
+ bd2 = 2
+ self.learn_hosts(bd1, lhosts)
+ self.learn_hosts(bd2, lhosts)
+ self.config_l2_fib_entries(bd1, hosts)
+ self.config_l2_fib_entries(bd2, hosts)
+ self.run_verify_test(bd1, lhosts, hosts)
+ self.run_verify_test(bd2, lhosts, hosts)
+
+ def test_l2_fib_flush_int(self):
+ """L2 FIB - flush interface"""
+ hosts = self.create_hosts(20, subnet=36)
+ lhosts = self.split_hosts(hosts, 10)
+
+ bd1 = 1
+ self.learn_hosts(bd1, lhosts)
+ self.config_l2_fib_entries(bd1, hosts)
+ self.run_verify_test(bd1, lhosts, hosts)
+ flushed = self.flush_int(self.pg_interfaces[0].sw_if_index, lhosts)
+ self.run_verify_test(bd1, hosts, lhosts)
+ self.run_verify_negat_test(bd1, hosts, flushed)
+
+ def test_l2_fib_flush_bd(self):
+ """L2 FIB - flush BD"""
+ hosts = self.create_hosts(20, subnet=37)
+ lhosts = self.split_hosts(hosts, 10)
+
+ bd1 = 1
+ self.learn_hosts(bd1, lhosts)
+ self.config_l2_fib_entries(bd1, hosts)
+ self.run_verify_test(bd1, lhosts, hosts)
+ flushed = self.flush_bd(bd1, lhosts)
+ self.run_verify_negat_test(bd1, hosts, flushed)
+
+ def test_l2_fib_flush_all(self):
+ """L2 FIB - flush all"""
+ hosts = self.create_hosts(20, subnet=38)
+ lhosts = self.split_hosts(hosts, 10)
+
+ bd1 = 1
+ bd2 = 2
+ self.learn_hosts(bd1, lhosts)
+ self.learn_hosts(bd2, lhosts)
+ self.config_l2_fib_entries(bd1, hosts)
+ self.config_l2_fib_entries(bd2, hosts)
+ self.run_verify_test(bd1, hosts, lhosts)
+ self.run_verify_test(bd2, hosts, lhosts)
+
+ self.flush_all()
+
+ self.run_verify_negat_test(bd1, hosts, lhosts)
+ self.run_verify_negat_test(bd2, hosts, lhosts)
+
+ def test_l2_fib_mac_learn_evs(self):
+ """L2 FIB - mac learning events"""
+ bd1 = 1
+ hosts = self.create_hosts(10, subnet=39)
+
+ self.vapi.want_l2_macs_events()
+ self.learn_hosts(bd1, hosts)
+
+ self.virtual_sleep(1)
+ self.logger.info(self.vapi.ppcli("show l2fib"))
+ 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]
+ }
+ self.vapi.want_l2_macs_events(enable_disable=0)
+ self.assertEqual(len(learned_macs ^ macs), 0)
- def test_l2_fib_03(self):
- """ L2 FIB test 3 - program new 100 MAC addresses
- """
- # Config 3
- # Create new test host entries
- self.create_hosts(100, subnet=22)
+ def test_l2_fib_mac_learn_evs2(self):
+ """L2 FIB - mac learning events using want_l2_macs_events2"""
+ bd1 = 1
+ hosts = self.create_hosts(10, subnet=39)
- # Add new 100 MAC entries to L2 FIB
- self.config_l2_fib_entries(bd_id=1, n_hosts_per_if=100)
+ self.vapi.l2fib_set_scan_delay(scan_delay=10)
+ self.vapi.want_l2_macs_events2()
+ self.sleep(1)
+ self.learn_hosts(bd1, hosts)
- # Test 3
- self.run_verify_test(bd_id=1)
+ self.virtual_sleep(1)
+ self.logger.info(self.vapi.ppcli("show l2fib"))
+ 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]
+ }
+ self.vapi.want_l2_macs_events2(enable_disable=0)
+ self.assertEqual(len(learned_macs ^ macs), 0)
- def test_l2_fib_04(self):
- """ L2 FIB test 4 - delete 160 MAC entries
- """
- # Config 4
- # Delete 160 MAC entries per interface from L2 FIB
- self.delete_l2_fib_entry(bd_id=1, n_hosts_per_if=160)
+ def test_l2_fib_macs_learn_max(self):
+ """L2 FIB - mac learning max macs in event"""
+ bd1 = 1
+ hosts = self.create_hosts(10, subnet=40)
- # Test 4a
- self.run_verify_negat_test(bd_id=1)
+ ev_macs = 1
+ self.vapi.want_l2_macs_events(max_macs_in_event=ev_macs)
+ self.learn_hosts(bd1, hosts)
- def test_l2_fib_05(self):
- """ L2 FIB test 5 - Program 10 new MAC entries, learn 10
- """
- self.create_hosts(20, subnet=35)
-
- self.learn_hosts(bd_id=1, n_hosts_per_if=10)
- self.learn_hosts(bd_id=2, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=1, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=2, n_hosts_per_if=10)
- self.run_verify_test(bd_id=1, dst_hosts=self.learned_hosts)
- self.run_verify_test(bd_id=2, dst_hosts=self.learned_hosts)
+ self.sleep(1)
+ self.logger.info(self.vapi.ppcli("show l2fib"))
+ evs = self.vapi.collect_events()
+ self.vapi.want_l2_macs_events(enable_disable=0)
- def test_l2_fib_06(self):
- """ L2 FIB test 6 - flush first interface
- """
- self.create_hosts(20, subnet=36)
-
- self.learn_hosts(bd_id=1, n_hosts_per_if=10)
- self.learn_hosts(bd_id=2, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=1, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=2, n_hosts_per_if=10)
- flushed = self.flush_int(self.pg_interfaces[0].sw_if_index)
- self.run_verify_test(bd_id=1, dst_hosts=self.learned_hosts)
- self.run_verify_negat_test(bd_id=1, dst_hosts=flushed)
-
- def test_l2_fib_07(self):
- """ L2 FIB test 7 - flush bd_id
- """
- self.create_hosts(20, subnet=37)
-
- self.learn_hosts(bd_id=1, n_hosts_per_if=10)
- self.learn_hosts(bd_id=2, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=1, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=2, n_hosts_per_if=10)
- flushed = self.flush_bd(bd_id=1)
- self.run_verify_negat_test(bd_id=1, dst_hosts=flushed)
- self.run_verify_test(bd_id=2, dst_hosts=self.learned_hosts)
-
- def test_l2_fib_08(self):
- """ L2 FIB test 8 - flush all
- """
- self.create_hosts(20, subnet=38)
-
- self.learn_hosts(bd_id=1, n_hosts_per_if=10)
- self.learn_hosts(bd_id=2, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=1, n_hosts_per_if=10)
- self.config_l2_fib_entries(bd_id=2, n_hosts_per_if=10)
- flushed = self.flush_all()
- self.run_verify_negat_test(bd_id=1, dst_hosts=flushed)
- self.run_verify_negat_test(bd_id=2, dst_hosts=flushed)
-
- def test_l2_fib_09(self):
- """ L2 FIB test 9 - mac learning events
- """
- self.create_hosts(10, subnet=39)
+ 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]
+ }
+
+ for e in evs:
+ self.assertLess(len(e), ev_macs * 10)
+ self.assertEqual(len(learned_macs ^ macs), 0)
- self.vapi.want_macs_learn_events()
- self.learn_hosts(bd_id=1, n_hosts_per_if=10)
+ def test_l2_fib_macs_learn_max2(self):
+ """L2 FIB - mac learning max macs in event using want_l2_macs_events2"""
+ bd1 = 1
+ hosts = self.create_hosts(10, subnet=40)
+ ev_macs = 1
+ self.vapi.l2fib_set_scan_delay(scan_delay=10)
+ self.vapi.want_l2_macs_events2(max_macs_in_event=ev_macs)
self.sleep(1)
+ self.learn_hosts(bd1, hosts)
+
+ self.virtual_sleep(1)
self.logger.info(self.vapi.ppcli("show l2fib"))
evs = self.vapi.collect_events()
+ self.vapi.want_l2_macs_events2(enable_disable=0)
+
+ 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 for e in evs for i in range(e.n_macs)}
- macs = {h.bin_mac for swif_hs in self.learned_hosts.itervalues()
- for h in swif_hs}
- self.vapi.want_macs_learn_events(enable_disable=0)
+ 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