3 from framework import VppTestCase
4 from ipaddress import IPv6Address
5 from scapy.contrib.gtp import *
6 from scapy.all import *
9 class TestSRv6uSIDSRH(VppTestCase):
10 """SRv6 End.uSID w/ SRH"""
14 super(TestSRv6uSIDSRH, cls).setUpClass()
16 cls.create_pg_interfaces(range(2))
17 cls.pg_if_i = cls.pg_interfaces[0]
18 cls.pg_if_o = cls.pg_interfaces[1]
20 cls.pg_if_i.config_ip6()
21 cls.pg_if_o.config_ip6()
23 cls.ip6_nhop = cls.pg_if_o.remote_ip6
25 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
26 cls.ip6_src = "1111:2222::1"
28 for pg_if in cls.pg_interfaces:
30 pg_if.resolve_ndp(timeout=5)
33 super(TestSRv6uSIDSRH, cls).tearDownClass()
36 def create_packets(self, inner):
37 ip6_dst = IPv6Address(str(self.ip6_dst))
39 ip6_src = IPv6Address(str(self.ip6_src))
41 self.logger.info("ip6 dst: {}".format(ip6_dst))
42 self.logger.info("ip6 src: {}".format(ip6_src))
48 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
49 / IPv6ExtHdrSegmentRouting(
53 addresses=["a1::1", "1111:2222:aaaa:bbbb::"],
56 / UDP(sport=1000, dport=23)
58 self.logger.info(pkt.show2(dump=True))
63 def test_srv6_usid_srh(self):
64 """test_srv6_usid_srh"""
65 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
67 self.vapi.cli("set sr encaps source addr A1::1")
68 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
69 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
71 self.logger.info(self.vapi.cli("show sr localsids"))
73 self.vapi.cli("clear errors")
75 self.pg0.add_stream(pkts)
76 self.pg_enable_capture(self.pg_interfaces)
79 self.logger.info(self.vapi.cli("show errors"))
80 self.logger.info(self.vapi.cli("show int address"))
82 capture = self.pg1.get_capture(len(pkts))
85 self.logger.info(pkt.show2(dump=True))
86 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
89 class TestSRv6uSID(VppTestCase):
90 """SRv6 End.uSID w/o SRH"""
94 super(TestSRv6uSID, cls).setUpClass()
96 cls.create_pg_interfaces(range(2))
97 cls.pg_if_i = cls.pg_interfaces[0]
98 cls.pg_if_o = cls.pg_interfaces[1]
100 cls.pg_if_i.config_ip6()
101 cls.pg_if_o.config_ip6()
103 cls.ip6_nhop = cls.pg_if_o.remote_ip6
105 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
106 cls.ip6_src = "1111:2222::1"
108 for pg_if in cls.pg_interfaces:
110 pg_if.resolve_ndp(timeout=5)
113 super(TestSRv6uSID, cls).tearDownClass()
116 def create_packets(self, inner):
117 ip6_dst = IPv6Address(str(self.ip6_dst))
119 ip6_src = IPv6Address(str(self.ip6_src))
121 self.logger.info("ip6 dst: {}".format(ip6_dst))
122 self.logger.info("ip6 src: {}".format(ip6_src))
128 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
130 / UDP(sport=1000, dport=23)
132 self.logger.info(pkt.show2(dump=True))
137 def test_srv6_usid(self):
139 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
141 self.vapi.cli("set sr encaps source addr A1::1")
142 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
143 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
145 self.logger.info(self.vapi.cli("show sr localsids"))
147 self.vapi.cli("clear errors")
149 self.pg0.add_stream(pkts)
150 self.pg_enable_capture(self.pg_interfaces)
153 self.logger.info(self.vapi.cli("show errors"))
154 self.logger.info(self.vapi.cli("show int address"))
156 capture = self.pg1.get_capture(len(pkts))
159 self.logger.info(pkt.show2(dump=True))
160 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
163 class TestSRv6uSIDFlexSRH(VppTestCase):
164 """SRv6 End.uSID.Flex w/ SRH"""
168 super(TestSRv6uSIDFlexSRH, cls).setUpClass()
170 cls.create_pg_interfaces(range(2))
171 cls.pg_if_i = cls.pg_interfaces[0]
172 cls.pg_if_o = cls.pg_interfaces[1]
174 cls.pg_if_i.config_ip6()
175 cls.pg_if_o.config_ip6()
177 cls.ip6_nhop = cls.pg_if_o.remote_ip6
179 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
180 cls.ip6_src = "1111:2222::1"
182 for pg_if in cls.pg_interfaces:
184 pg_if.resolve_ndp(timeout=5)
187 super(TestSRv6uSIDFlexSRH, cls).tearDownClass()
190 def create_packets(self, inner):
191 ip6_dst = IPv6Address(str(self.ip6_dst))
193 ip6_src = IPv6Address(str(self.ip6_src))
195 self.logger.info("ip6 dst: {}".format(ip6_dst))
196 self.logger.info("ip6 src: {}".format(ip6_src))
202 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
203 / IPv6ExtHdrSegmentRouting(
207 addresses=["a1::1", "1111:2222:aaaa:bbbb::"],
210 / UDP(sport=1000, dport=23)
212 self.logger.info(pkt.show2(dump=True))
217 def test_srv6_usid_flex_srh(self):
218 """test_srv6_usid_flex_srh"""
219 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
221 self.vapi.cli("set sr encaps source addr A1::1")
222 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
223 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
225 self.logger.info(self.vapi.cli("show sr localsids"))
227 self.vapi.cli("clear errors")
229 self.pg0.add_stream(pkts)
230 self.pg_enable_capture(self.pg_interfaces)
233 self.logger.info(self.vapi.cli("show errors"))
234 self.logger.info(self.vapi.cli("show int address"))
236 capture = self.pg1.get_capture(len(pkts))
239 self.logger.info(pkt.show2(dump=True))
240 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
243 class TestSRv6uSIDFlex(VppTestCase):
244 """SRv6 End.uSID.Flex w/o SRH"""
248 super(TestSRv6uSIDFlex, cls).setUpClass()
250 cls.create_pg_interfaces(range(2))
251 cls.pg_if_i = cls.pg_interfaces[0]
252 cls.pg_if_o = cls.pg_interfaces[1]
254 cls.pg_if_i.config_ip6()
255 cls.pg_if_o.config_ip6()
257 cls.ip6_nhop = cls.pg_if_o.remote_ip6
259 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
260 cls.ip6_src = "1111:2222::1"
262 for pg_if in cls.pg_interfaces:
264 pg_if.resolve_ndp(timeout=5)
267 super(TestSRv6uSIDFlex, cls).tearDownClass()
270 def create_packets(self, inner):
271 ip6_dst = IPv6Address(str(self.ip6_dst))
273 ip6_src = IPv6Address(str(self.ip6_src))
275 self.logger.info("ip6 dst: {}".format(ip6_dst))
276 self.logger.info("ip6 src: {}".format(ip6_src))
282 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
284 / UDP(sport=1000, dport=23)
286 self.logger.info(pkt.show2(dump=True))
291 def test_srv6_usid_flex(self):
292 """test_srv6_usid_flex"""
293 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
295 self.vapi.cli("set sr encaps source addr A1::1")
296 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
297 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
299 self.logger.info(self.vapi.cli("show sr localsids"))
301 self.vapi.cli("clear errors")
303 self.pg0.add_stream(pkts)
304 self.pg_enable_capture(self.pg_interfaces)
307 self.logger.info(self.vapi.cli("show errors"))
308 self.logger.info(self.vapi.cli("show int address"))
310 capture = self.pg1.get_capture(len(pkts))
313 self.logger.info(pkt.show2(dump=True))
314 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")