3 from framework import VppTestCase
4 from ipaddress import IPv4Address
5 from ipaddress import IPv6Address
6 from scapy.contrib.gtp import *
7 from scapy.all import *
10 class TestSRv6uSIDSRH(VppTestCase):
11 """SRv6 End.uSID w/ SRH"""
15 super(TestSRv6uSIDSRH, cls).setUpClass()
17 cls.create_pg_interfaces(range(2))
18 cls.pg_if_i = cls.pg_interfaces[0]
19 cls.pg_if_o = cls.pg_interfaces[1]
21 cls.pg_if_i.config_ip6()
22 cls.pg_if_o.config_ip6()
24 cls.ip6_nhop = cls.pg_if_o.remote_ip6
26 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
27 cls.ip6_src = "1111:2222::1"
29 for pg_if in cls.pg_interfaces:
31 pg_if.resolve_ndp(timeout=5)
34 super(TestSRv6uSIDSRH, cls).tearDownClass()
37 def create_packets(self, inner):
38 ip6_dst = IPv6Address(str(self.ip6_dst))
40 ip6_src = IPv6Address(str(self.ip6_src))
42 self.logger.info("ip6 dst: {}".format(ip6_dst))
43 self.logger.info("ip6 src: {}".format(ip6_src))
49 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
50 / IPv6ExtHdrSegmentRouting(
54 addresses=["a1::1", "1111:2222:aaaa:bbbb::"],
57 / UDP(sport=1000, dport=23)
59 self.logger.info(pkt.show2(dump=True))
64 def test_srv6_usid_srh(self):
65 """test_srv6_usid_srh"""
66 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
68 self.vapi.cli("set sr encaps source addr A1::1")
69 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
70 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
72 self.logger.info(self.vapi.cli("show sr localsids"))
74 self.vapi.cli("clear errors")
76 self.pg0.add_stream(pkts)
77 self.pg_enable_capture(self.pg_interfaces)
80 self.logger.info(self.vapi.cli("show errors"))
81 self.logger.info(self.vapi.cli("show int address"))
83 capture = self.pg1.get_capture(len(pkts))
86 self.logger.info(pkt.show2(dump=True))
87 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
90 class TestSRv6uSID(VppTestCase):
91 """SRv6 End.uSID w/o SRH"""
95 super(TestSRv6uSID, cls).setUpClass()
97 cls.create_pg_interfaces(range(2))
98 cls.pg_if_i = cls.pg_interfaces[0]
99 cls.pg_if_o = cls.pg_interfaces[1]
101 cls.pg_if_i.config_ip6()
102 cls.pg_if_o.config_ip6()
104 cls.ip6_nhop = cls.pg_if_o.remote_ip6
106 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
107 cls.ip6_src = "1111:2222::1"
109 for pg_if in cls.pg_interfaces:
111 pg_if.resolve_ndp(timeout=5)
114 super(TestSRv6uSID, cls).tearDownClass()
117 def create_packets(self, inner):
118 ip6_dst = IPv6Address(str(self.ip6_dst))
120 ip6_src = IPv6Address(str(self.ip6_src))
122 self.logger.info("ip6 dst: {}".format(ip6_dst))
123 self.logger.info("ip6 src: {}".format(ip6_src))
129 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
131 / UDP(sport=1000, dport=23)
133 self.logger.info(pkt.show2(dump=True))
138 def test_srv6_usid(self):
140 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
142 self.vapi.cli("set sr encaps source addr A1::1")
143 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
144 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
146 self.logger.info(self.vapi.cli("show sr localsids"))
148 self.vapi.cli("clear errors")
150 self.pg0.add_stream(pkts)
151 self.pg_enable_capture(self.pg_interfaces)
154 self.logger.info(self.vapi.cli("show errors"))
155 self.logger.info(self.vapi.cli("show int address"))
157 capture = self.pg1.get_capture(len(pkts))
160 self.logger.info(pkt.show2(dump=True))
161 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
164 class TestSRv6uSIDFlexSRH(VppTestCase):
165 """SRv6 End.uSID.Flex w/ SRH"""
169 super(TestSRv6uSIDFlexSRH, cls).setUpClass()
171 cls.create_pg_interfaces(range(2))
172 cls.pg_if_i = cls.pg_interfaces[0]
173 cls.pg_if_o = cls.pg_interfaces[1]
175 cls.pg_if_i.config_ip6()
176 cls.pg_if_o.config_ip6()
178 cls.ip6_nhop = cls.pg_if_o.remote_ip6
180 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
181 cls.ip6_src = "1111:2222::1"
183 for pg_if in cls.pg_interfaces:
185 pg_if.resolve_ndp(timeout=5)
188 super(TestSRv6uSIDFlexSRH, cls).tearDownClass()
191 def create_packets(self, inner):
192 ip6_dst = IPv6Address(str(self.ip6_dst))
194 ip6_src = IPv6Address(str(self.ip6_src))
196 self.logger.info("ip6 dst: {}".format(ip6_dst))
197 self.logger.info("ip6 src: {}".format(ip6_src))
203 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
204 / IPv6ExtHdrSegmentRouting(
208 addresses=["a1::1", "1111:2222:aaaa:bbbb::"],
211 / UDP(sport=1000, dport=23)
213 self.logger.info(pkt.show2(dump=True))
218 def test_srv6_usid_flex_srh(self):
219 """test_srv6_usid_flex_srh"""
220 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
222 self.vapi.cli("set sr encaps source addr A1::1")
223 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
224 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
226 self.logger.info(self.vapi.cli("show sr localsids"))
228 self.vapi.cli("clear errors")
230 self.pg0.add_stream(pkts)
231 self.pg_enable_capture(self.pg_interfaces)
234 self.logger.info(self.vapi.cli("show errors"))
235 self.logger.info(self.vapi.cli("show int address"))
237 capture = self.pg1.get_capture(len(pkts))
240 self.logger.info(pkt.show2(dump=True))
241 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
244 class TestSRv6uSIDFlex(VppTestCase):
245 """SRv6 End.uSID.Flex w/o SRH"""
249 super(TestSRv6uSIDFlex, cls).setUpClass()
251 cls.create_pg_interfaces(range(2))
252 cls.pg_if_i = cls.pg_interfaces[0]
253 cls.pg_if_o = cls.pg_interfaces[1]
255 cls.pg_if_i.config_ip6()
256 cls.pg_if_o.config_ip6()
258 cls.ip6_nhop = cls.pg_if_o.remote_ip6
260 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
261 cls.ip6_src = "1111:2222::1"
263 for pg_if in cls.pg_interfaces:
265 pg_if.resolve_ndp(timeout=5)
268 super(TestSRv6uSIDFlex, cls).tearDownClass()
271 def create_packets(self, inner):
272 ip6_dst = IPv6Address(str(self.ip6_dst))
274 ip6_src = IPv6Address(str(self.ip6_src))
276 self.logger.info("ip6 dst: {}".format(ip6_dst))
277 self.logger.info("ip6 src: {}".format(ip6_src))
283 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
285 / UDP(sport=1000, dport=23)
287 self.logger.info(pkt.show2(dump=True))
292 def test_srv6_usid_flex(self):
293 """test_srv6_usid_flex"""
294 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
296 self.vapi.cli("set sr encaps source addr A1::1")
297 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
298 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
300 self.logger.info(self.vapi.cli("show sr localsids"))
302 self.vapi.cli("clear errors")
304 self.pg0.add_stream(pkts)
305 self.pg_enable_capture(self.pg_interfaces)
308 self.logger.info(self.vapi.cli("show errors"))
309 self.logger.info(self.vapi.cli("show int address"))
311 capture = self.pg1.get_capture(len(pkts))
314 self.logger.info(pkt.show2(dump=True))
315 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")