diff options
Diffstat (limited to 'test/test_l2_fib.py')
| -rw-r--r-- | test/test_l2_fib.py | 172 |
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) |