X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_l3xc.py;h=66eb242ff52800d1a68301524d6f68f3f7b2cd5f;hb=6ee3aa41c395d036c8c79a3681cc5ee6bc6fceb9;hp=ceb95ce02dde3524aee561ea029ac59608239d42;hpb=59fa121f8953f7b07f0cc02149ca28182f959f42;p=vpp.git diff --git a/test/test_l3xc.py b/test/test_l3xc.py index ceb95ce02dd..66eb242ff52 100644 --- a/test/test_l3xc.py +++ b/test/test_l3xc.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6 import unittest @@ -28,61 +28,38 @@ def find_l3xc(test, sw_if_index, dump_sw_if_index=None): class VppL3xc(VppObject): - - def __init__(self, test, intf, paths, is_ip6=False): + def __init__(self, test, intf, paths, is_ip6=False): self._test = test self.intf = intf self.is_ip6 = is_ip6 self.paths = paths - - def encode_paths(self): - br_paths = [] - for p in self.paths: - lstack = [] - for l in p.nh_labels: - if type(l) == VppMplsLabel: - lstack.append(l.encode()) - else: - lstack.append({'label': l, 'ttl': 255}) - n_labels = len(lstack) - while (len(lstack) < 16): - lstack.append({}) - br_paths.append({'next_hop': p.nh_addr, - 'weight': 1, - 'afi': p.proto, - 'sw_if_index': p.nh_itf, - 'preference': 0, - 'table_id': p.nh_table_id, - 'next_hop_id': p.next_hop_id, - 'is_udp_encap': p.is_udp_encap, - 'n_labels': n_labels, - 'label_stack': lstack}) - return br_paths + self.encoded_paths = [] + for path in self.paths: + self.encoded_paths.append(path.encode()) def add_vpp_config(self): self._test.vapi.l3xc_update( l3xc={ - 'is_ip6': self.is_ip6, - 'sw_if_index': self.intf.sw_if_index, - 'n_paths': len(self.paths), - 'paths': self.encode_paths() - }) + "is_ip6": self.is_ip6, + "sw_if_index": self.intf.sw_if_index, + "n_paths": len(self.paths), + "paths": self.encoded_paths, + } + ) self._test.registry.register(self, self._test.logger) def remove_vpp_config(self): - self._test.vapi.l3xc_del( - is_ip6=self.is_ip6, - sw_if_index=self.intf.sw_if_index) + self._test.vapi.l3xc_del(is_ip6=self.is_ip6, sw_if_index=self.intf.sw_if_index) def query_vpp_config(self): return find_l3xc(self._test, self.intf.sw_if_index) def object_id(self): - return ("l3xc-%d" % self.intf.sw_if_index) + return "l3xc-%d" % self.intf.sw_if_index class TestL3xc(VppTestCase): - """ L3XC Test Case """ + """L3XC Test Case""" @classmethod def setUpClass(cls): @@ -108,40 +85,31 @@ class TestL3xc(VppTestCase): for i in self.pg_interfaces: i.unconfig_ip4() i.unconfig_ip6() - i.ip6_disable() i.admin_down() super(TestL3xc, self).tearDown() - def send_and_expect_load_balancing(self, input, pkts, outputs): - self.pg_send(input, pkts) - rxs = [] - for oo in outputs: - rx = oo._get_capture(1) - self.assertNotEqual(0, len(rx)) - for r in rx: - rxs.append(r) - return rxs - def test_l3xc4(self): - """ IPv4 X-Connect """ + """IPv4 X-Connect""" # # x-connect pg0 to pg1 and pg2 to pg3->5 # - l3xc_1 = VppL3xc(self, self.pg0, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) + l3xc_1 = VppL3xc( + self, self.pg0, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)] + ) l3xc_1.add_vpp_config() - l3xc_2 = VppL3xc(self, self.pg2, - [VppRoutePath(self.pg3.remote_ip4, - self.pg3.sw_if_index), - VppRoutePath(self.pg4.remote_ip4, - self.pg4.sw_if_index), - VppRoutePath(self.pg5.remote_ip4, - self.pg5.sw_if_index)]) + l3xc_2 = VppL3xc( + self, + self.pg2, + [ + VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index), + VppRoutePath(self.pg4.remote_ip4, self.pg4.sw_if_index), + VppRoutePath(self.pg5.remote_ip4, self.pg5.sw_if_index), + ], + ) l3xc_2.add_vpp_config() - self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xffffffff)) + self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xFFFFFFFF)) self.logger.info(self.vapi.cli("sh l3xc")) @@ -149,26 +117,29 @@ class TestL3xc(VppTestCase): # fire in packets. If it's forwarded then the L3XC was successful, # since default routing will drop it # - p_1 = (Ether(src=self.pg0.remote_mac, - dst=self.pg0.local_mac) / - IP(src="1.1.1.1", dst="1.1.1.2") / - UDP(sport=1234, dport=1234) / - Raw('\xa5' * 100)) + p_1 = ( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src="1.1.1.1", dst="1.1.1.2") + / UDP(sport=1234, dport=1234) + / Raw(b"\xa5" * 100) + ) # self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1) p_2 = [] for ii in range(NUM_PKTS): - p_2.append(Ether(src=self.pg0.remote_mac, - dst=self.pg0.local_mac) / - IP(src="1.1.1.1", dst="1.1.1.2") / - UDP(sport=1000 + ii, dport=1234) / - Raw('\xa5' * 100)) - self.send_and_expect_load_balancing(self.pg2, p_2, - [self.pg3, self.pg4, self.pg5]) + p_2.append( + Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) + / IP(src="1.1.1.1", dst="1.1.1.2") + / UDP(sport=1000 + ii, dport=1234) + / Raw(b"\xa5" * 100) + ) + self.send_and_expect_load_balancing( + self.pg2, p_2, [self.pg3, self.pg4, self.pg5] + ) l3xc_2.remove_vpp_config() self.send_and_assert_no_replies(self.pg2, p_2) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)