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):
39 ip6_dst = IPv6Address(str(self.ip6_dst))
41 ip6_src = IPv6Address(str(self.ip6_src))
43 self.logger.info("ip6 dst: {}".format(ip6_dst))
44 self.logger.info("ip6 src: {}".format(ip6_src))
50 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
51 / IPv6ExtHdrSegmentRouting(
55 addresses=["a1::1", "1111:2222:aaaa:bbbb::"],
58 / UDP(sport=1000, dport=23)
60 self.logger.info(pkt.show2(dump=True))
65 def test_srv6_usid_srh(self):
66 """test_srv6_usid_srh"""
67 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
69 self.vapi.cli("set sr encaps source addr A1::1")
70 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
71 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
73 self.logger.info(self.vapi.cli("show sr localsids"))
75 self.vapi.cli("clear errors")
77 self.pg0.add_stream(pkts)
78 self.pg_enable_capture(self.pg_interfaces)
81 self.logger.info(self.vapi.cli("show errors"))
82 self.logger.info(self.vapi.cli("show int address"))
84 capture = self.pg1.get_capture(len(pkts))
87 self.logger.info(pkt.show2(dump=True))
88 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
91 class TestSRv6uSID(VppTestCase):
92 """SRv6 End.uSID w/o SRH"""
96 super(TestSRv6uSID, cls).setUpClass()
98 cls.create_pg_interfaces(range(2))
99 cls.pg_if_i = cls.pg_interfaces[0]
100 cls.pg_if_o = cls.pg_interfaces[1]
102 cls.pg_if_i.config_ip6()
103 cls.pg_if_o.config_ip6()
105 cls.ip6_nhop = cls.pg_if_o.remote_ip6
107 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
108 cls.ip6_src = "1111:2222::1"
110 for pg_if in cls.pg_interfaces:
112 pg_if.resolve_ndp(timeout=5)
115 super(TestSRv6uSID, cls).tearDownClass()
118 def create_packets(self, inner):
120 ip6_dst = IPv6Address(str(self.ip6_dst))
122 ip6_src = IPv6Address(str(self.ip6_src))
124 self.logger.info("ip6 dst: {}".format(ip6_dst))
125 self.logger.info("ip6 src: {}".format(ip6_src))
131 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
133 / UDP(sport=1000, dport=23)
135 self.logger.info(pkt.show2(dump=True))
140 def test_srv6_usid(self):
142 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
144 self.vapi.cli("set sr encaps source addr A1::1")
145 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
146 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
148 self.logger.info(self.vapi.cli("show sr localsids"))
150 self.vapi.cli("clear errors")
152 self.pg0.add_stream(pkts)
153 self.pg_enable_capture(self.pg_interfaces)
156 self.logger.info(self.vapi.cli("show errors"))
157 self.logger.info(self.vapi.cli("show int address"))
159 capture = self.pg1.get_capture(len(pkts))
162 self.logger.info(pkt.show2(dump=True))
163 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
166 class TestSRv6uSIDFlexSRH(VppTestCase):
167 """SRv6 End.uSID.Flex w/ SRH"""
171 super(TestSRv6uSIDFlexSRH, cls).setUpClass()
173 cls.create_pg_interfaces(range(2))
174 cls.pg_if_i = cls.pg_interfaces[0]
175 cls.pg_if_o = cls.pg_interfaces[1]
177 cls.pg_if_i.config_ip6()
178 cls.pg_if_o.config_ip6()
180 cls.ip6_nhop = cls.pg_if_o.remote_ip6
182 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
183 cls.ip6_src = "1111:2222::1"
185 for pg_if in cls.pg_interfaces:
187 pg_if.resolve_ndp(timeout=5)
190 super(TestSRv6uSIDFlexSRH, cls).tearDownClass()
193 def create_packets(self, inner):
195 ip6_dst = IPv6Address(str(self.ip6_dst))
197 ip6_src = IPv6Address(str(self.ip6_src))
199 self.logger.info("ip6 dst: {}".format(ip6_dst))
200 self.logger.info("ip6 src: {}".format(ip6_src))
206 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
207 / IPv6ExtHdrSegmentRouting(
211 addresses=["a1::1", "1111:2222:aaaa:bbbb::"],
214 / UDP(sport=1000, dport=23)
216 self.logger.info(pkt.show2(dump=True))
221 def test_srv6_usid_flex_srh(self):
222 """test_srv6_usid_flex_srh"""
223 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
225 self.vapi.cli("set sr encaps source addr A1::1")
226 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
227 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
229 self.logger.info(self.vapi.cli("show sr localsids"))
231 self.vapi.cli("clear errors")
233 self.pg0.add_stream(pkts)
234 self.pg_enable_capture(self.pg_interfaces)
237 self.logger.info(self.vapi.cli("show errors"))
238 self.logger.info(self.vapi.cli("show int address"))
240 capture = self.pg1.get_capture(len(pkts))
243 self.logger.info(pkt.show2(dump=True))
244 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
247 class TestSRv6uSIDFlex(VppTestCase):
248 """SRv6 End.uSID.Flex w/o SRH"""
252 super(TestSRv6uSIDFlex, cls).setUpClass()
254 cls.create_pg_interfaces(range(2))
255 cls.pg_if_i = cls.pg_interfaces[0]
256 cls.pg_if_o = cls.pg_interfaces[1]
258 cls.pg_if_i.config_ip6()
259 cls.pg_if_o.config_ip6()
261 cls.ip6_nhop = cls.pg_if_o.remote_ip6
263 cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
264 cls.ip6_src = "1111:2222::1"
266 for pg_if in cls.pg_interfaces:
268 pg_if.resolve_ndp(timeout=5)
271 super(TestSRv6uSIDFlex, cls).tearDownClass()
274 def create_packets(self, inner):
276 ip6_dst = IPv6Address(str(self.ip6_dst))
278 ip6_src = IPv6Address(str(self.ip6_src))
280 self.logger.info("ip6 dst: {}".format(ip6_dst))
281 self.logger.info("ip6 src: {}".format(ip6_src))
287 / IPv6(dst=str(ip6_dst), src=str(ip6_src))
289 / UDP(sport=1000, dport=23)
291 self.logger.info(pkt.show2(dump=True))
296 def test_srv6_usid_flex(self):
297 """test_srv6_usid_flex"""
298 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
300 self.vapi.cli("set sr encaps source addr A1::1")
301 self.vapi.cli("sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
302 self.vapi.cli("ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
304 self.logger.info(self.vapi.cli("show sr localsids"))
306 self.vapi.cli("clear errors")
308 self.pg0.add_stream(pkts)
309 self.pg_enable_capture(self.pg_interfaces)
312 self.logger.info(self.vapi.cli("show errors"))
313 self.logger.info(self.vapi.cli("show int address"))
315 capture = self.pg1.get_capture(len(pkts))
318 self.logger.info(pkt.show2(dump=True))
319 self.assertEqual(pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")