-#!/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 src/vnet/l2/l2_fib.h
-MAC_EVENT_ACTION_ADD = 0
-MAC_EVENT_ACTION_DELETE = 1
-MAC_EVENT_ACTION_MOVE = 2
+from vpp_papi import mac_pton, VppEnum
class TestL2fib(VppTestCase):
@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()
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:
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))
+ def show_commands_at_teardown(self):
+ self.logger.info(self.vapi.ppcli("show l2fib verbose"))
+
def create_hosts(self, n_hosts_per_if, subnet):
"""
Create required number of host MAC addresses and distribute them among
:param int bd_id: BD to teach
: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
swif = pg_if.sw_if_index
for host in hosts[swif]:
self.vapi.l2fib_add_del(
- host.mac, bd_id, swif, static_mac=1)
+ mac_pton(host.mac), bd_id, swif, static_mac=1)
def delete_l2_fib_entry(self, bd_id, hosts):
"""
swif = pg_if.sw_if_index
for host in hosts[swif]:
self.vapi.l2fib_add_del(
- host.mac, bd_id, swif, is_add=0)
+ mac_pton(host.mac), bd_id, swif, is_add=0)
def flush_int(self, swif, learned_hosts):
"""
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]
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:
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()
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.config_l2_fib_entries(bd_id, hosts)
self.run_verify_test(bd_id, hosts, hosts)
- def test_l2_fib_delete12(self):
- """ L2 FIB - program 100 + delete 12 MACs
+ 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.run_verify_test(bd_id, hosts, hosts)
self.run_verify_negat_test(bd_id, hosts, del_hosts)
- def test_l2_fib_add100_add100(self):
- """ L2 FIB - program 100 + 100 MACs
+ 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.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)
bd1 = 1
hosts = self.create_hosts(10, subnet=39)
- self.vapi.want_macs_learn_events()
+ self.vapi.want_l2_macs_events()
+ self.learn_hosts(bd1, hosts)
+
+ self.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_mac_learn_evs2(self):
+ """ L2 FIB - mac learning events using want_l2_macs_events2
+ """
+ bd1 = 1
+ hosts = self.create_hosts(10, subnet=39)
+
+ self.vapi.l2fib_set_scan_delay(scan_delay=10)
+ self.vapi.want_l2_macs_events2()
+ self.sleep(1)
self.learn_hosts(bd1, hosts)
self.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 for e in evs for i in range(e.n_macs)
- if e.mac[i].action == MAC_EVENT_ACTION_ADD}
+ 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_macs_learn_events(enable_disable=0)
+ self.vapi.want_l2_macs_events2(enable_disable=0)
self.assertEqual(len(learned_macs ^ macs), 0)
def test_l2_fib_macs_learn_max(self):
hosts = self.create_hosts(10, subnet=40)
ev_macs = 1
- self.vapi.want_macs_learn_events(max_macs_in_event=ev_macs)
+ self.vapi.want_l2_macs_events(max_macs_in_event=ev_macs)
+ self.learn_hosts(bd1, 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)
+
+ 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)
+
+ 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.sleep(1)
self.logger.info(self.vapi.ppcli("show l2fib"))
evs = self.vapi.collect_events()
- self.vapi.want_macs_learn_events(enable_disable=0)
+ 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)
- if e.mac[i].action == MAC_EVENT_ACTION_ADD}
+ 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]}
Code Review / vpp.git / blobdiff