summaryrefslogtreecommitdiffstats
path: root/test/test_l2_fib.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_l2_fib.py')
-rw-r--r--test/test_l2_fib.py172
1 files changed, 100 insertions, 72 deletions
diff --git a/test/test_l2_fib.py b/test/test_l2_fib.py
index 065e57e5343..fe1ea458182 100644
--- a/test/test_l2_fib.py
+++ b/test/test_l2_fib.py
@@ -73,12 +73,11 @@ 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 - 1)
+ return range((bd_id - 1) * cls.n_ifs_per_bd, bd_id * cls.n_ifs_per_bd - 1)
@classmethod
def setUpClass(cls):
@@ -104,7 +103,8 @@ class TestL2fib(VppTestCase):
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]
@@ -112,12 +112,12 @@ class TestL2fib(VppTestCase):
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:
@@ -141,8 +141,9 @@ class TestL2fib(VppTestCase):
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"))
@@ -160,11 +161,15 @@ class TestL2fib(VppTestCase):
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
@@ -188,8 +193,9 @@ class TestL2fib(VppTestCase):
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()
@@ -206,8 +212,7 @@ class TestL2fib(VppTestCase):
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):
"""
@@ -219,8 +224,7 @@ class TestL2fib(VppTestCase):
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):
"""
@@ -278,10 +282,12 @@ class TestL2fib(VppTestCase):
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)
@@ -306,11 +312,13 @@ class TestL2fib(VppTestCase):
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)
@@ -325,11 +333,13 @@ class TestL2fib(VppTestCase):
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
@@ -338,9 +348,11 @@ class TestL2fib(VppTestCase):
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)
@@ -366,9 +378,11 @@ class TestL2fib(VppTestCase):
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)
@@ -386,16 +400,14 @@ class TestL2fib(VppTestCase):
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)
@@ -406,8 +418,7 @@ class TestL2fib(VppTestCase):
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)
@@ -416,8 +427,7 @@ class TestL2fib(VppTestCase):
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)
@@ -431,8 +441,7 @@ class TestL2fib(VppTestCase):
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)
@@ -445,8 +454,7 @@ class TestL2fib(VppTestCase):
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)
@@ -458,8 +466,7 @@ class TestL2fib(VppTestCase):
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)
@@ -478,8 +485,7 @@ class TestL2fib(VppTestCase):
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)
@@ -491,16 +497,21 @@ class TestL2fib(VppTestCase):
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)
@@ -514,16 +525,21 @@ class TestL2fib(VppTestCase):
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)
@@ -539,18 +555,23 @@ class TestL2fib(VppTestCase):
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)
@@ -568,14 +589,21 @@ class TestL2fib(VppTestCase):
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)