5 from framework import VppTestCase
6 from asfframework import VppTestRunner
7 from vpp_ip_route import VppRoutePath
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether
11 from scapy.layers.inet import IP, UDP
13 from vpp_object import VppObject
18 def find_l3xc(test, sw_if_index, dump_sw_if_index=None):
19 if not dump_sw_if_index:
20 dump_sw_if_index = sw_if_index
21 xcs = test.vapi.l3xc_dump(dump_sw_if_index)
23 if sw_if_index == xc.l3xc.sw_if_index:
28 class VppL3xc(VppObject):
29 def __init__(self, test, intf, paths, is_ip6=False):
34 self.encoded_paths = []
35 for path in self.paths:
36 self.encoded_paths.append(path.encode())
38 def add_vpp_config(self):
39 self._test.vapi.l3xc_update(
41 "is_ip6": self.is_ip6,
42 "sw_if_index": self.intf.sw_if_index,
43 "n_paths": len(self.paths),
44 "paths": self.encoded_paths,
47 self._test.registry.register(self, self._test.logger)
49 def remove_vpp_config(self):
50 self._test.vapi.l3xc_del(is_ip6=self.is_ip6, sw_if_index=self.intf.sw_if_index)
52 def query_vpp_config(self):
53 return find_l3xc(self._test, self.intf.sw_if_index)
56 return "l3xc-%d" % self.intf.sw_if_index
59 class TestL3xc(VppTestCase):
64 super(TestL3xc, cls).setUpClass()
67 def tearDownClass(cls):
68 super(TestL3xc, cls).tearDownClass()
71 super(TestL3xc, self).setUp()
73 self.create_pg_interfaces(range(6))
75 for i in self.pg_interfaces:
83 for i in self.pg_interfaces:
87 super(TestL3xc, self).tearDown()
93 # x-connect pg0 to pg1 and pg2 to pg3->5
96 self, self.pg0, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)]
98 l3xc_1.add_vpp_config()
103 VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index),
104 VppRoutePath(self.pg4.remote_ip4, self.pg4.sw_if_index),
105 VppRoutePath(self.pg5.remote_ip4, self.pg5.sw_if_index),
108 l3xc_2.add_vpp_config()
110 self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xFFFFFFFF))
112 self.logger.info(self.vapi.cli("sh l3xc"))
115 # fire in packets. If it's forwarded then the L3XC was successful,
116 # since default routing will drop it
119 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
120 / IP(src="1.1.1.1", dst="1.1.1.2")
121 / UDP(sport=1234, dport=1234)
124 # self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1)
127 for ii in range(NUM_PKTS):
129 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
130 / IP(src="1.1.1.1", dst="1.1.1.2")
131 / UDP(sport=1000 + ii, dport=1234)
134 self.send_and_expect_load_balancing(
135 self.pg2, p_2, [self.pg3, self.pg4, self.pg5]
138 l3xc_2.remove_vpp_config()
139 self.send_and_assert_no_replies(self.pg2, p_2)
142 if __name__ == "__main__":
143 unittest.main(testRunner=VppTestRunner)