#!/usr/bin/env python """L2 FIB Test Case HLD: **config 1** - add 4 pg-l2 interfaces - configure them into l2bd - configure 100 MAC entries in L2 fib - 25 MACs per interface - L2 MAC learning and unknown unicast flooding disabled in l2bd - configure 100 MAC entries in L2 fib - 25 MACs per interface **test 1** - send L2 MAC frames between all 4 pg-l2 interfaces for all of 100 MAC \ entries in the FIB **verify 1** - all packets received correctly **config 2** - delete 12 MAC entries - 3 MACs per interface **test 2a** - send L2 MAC frames between all 4 pg-l2 interfaces for non-deleted MAC \ entries **verify 2a** - all packets received correctly **test 2b** - send L2 MAC frames between all 4 pg-l2 interfaces for all of 12 deleted \ MAC entries **verify 2b** - no packet received on all 4 pg-l2 interfaces **config 3** - configure new 100 MAC entries in L2 fib - 25 MACs per interface **test 3** - send L2 MAC frames between all 4 pg-l2 interfaces for all of 188 MAC \ entries in the FIB **verify 3** - all packets received correctly **config 4** - delete 160 MAC entries, 40 MACs per interface **test 4a** - send L2 MAC frames between all 4 pg-l2 interfaces for all of 28 \ non-deleted MAC entries **verify 4a** - all packets received correctly **test 4b** - try send L2 MAC frames between all 4 pg-l2 interfaces for all of 172 \ deleted MAC entries **verify 4b** - no packet received on all 4 pg-l2 interfaces """ import unittest import random from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP 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 class TestL2fib(VppTestCase): """ 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) @classmethod def setUpClass(cls): """ Perform standard class setup (defined by class method setUpClass in class VppTestCase) before running the test case, set test case related variables and configure VPP. :var int bd_id: Bridge domain ID. """ super(TestL2fib, cls).setUpClass() try: 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 pg interfaces cls.create_pg_interfaces(n_ifs) cls.flows = dict() for bd_id in n_brs: # Packet flows mapping pg0 -> pg1, pg2, pg3 etc. 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] # 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) 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) # Set up all interfaces for i in cls.pg_interfaces: i.admin_up() except Exception: super(TestL2fib, cls).tearDownClass() raise def setUp(self): super(TestL2fib, self).setUp() self.reset_packet_infos() def tearDown(self): """ Show various debug prints after each test. """ 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 create_hosts(self, n_hosts_per_if, subnet): """ Create required number of host MAC addresses and distribute them among 
# Copyright (c) 2017-2019 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# Find uncrustify program
#
find_program( UNCRUSTIFY_BIN uncrustify
                PATHS
				$ENV{UNCRUSTIFY_HOME}
				)

message( "-- UNCRUSTIFY found in ${UNCRUSTIFY_BIN}" )
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_macs_learn_events() self.learn_hosts(bd1, hosts) self.sleep(1) self.logger.info(self.vapi.ppcli("show l2fib")) evs = self.vapi.collect_events() 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} 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.assertEqual(len(learned_macs ^ macs), 0) 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) ev_macs = 1 self.vapi.want_macs_learn_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_macs_learn_events(enable_disable=0) self.assertGreater(len(evs), 0) 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} 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__': unittest.main(testRunner=VppTestRunner)