7 from framework import VppTestCase, VppTestRunner
8 from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
10 from scapy.packet import Raw
11 from scapy.layers.l2 import Ether, Dot1Q, ARP
12 from scapy.layers.inet import IP, UDP
13 from scapy.layers.inet6 import ICMPv6ND_NS, IPv6, UDP
14 from scapy.contrib.mpls import MPLS
16 class TestMPLS(VppTestCase):
17 """ MPLS Test Case """
21 super(TestMPLS, cls).setUpClass()
24 super(TestMPLS, self).setUp()
26 # create 2 pg interfaces
27 self.create_pg_interfaces(range(3))
29 # setup both interfaces
30 # assign them different tables.
33 for i in self.pg_interfaces:
35 i.set_table_ip4(table_id)
36 i.set_table_ip6(table_id)
45 super(TestMPLS, self).tearDown()
47 def create_stream_ip4(self, src_if, mpls_label, mpls_ttl):
49 for i in range(0, 257):
50 info = self.create_packet_info(src_if.sw_if_index,
52 payload = self.info_to_payload(info)
53 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
54 MPLS(label=mpls_label, ttl=mpls_ttl) /
55 IP(src=src_if.remote_ip4, dst=src_if.remote_ip4) /
56 UDP(sport=1234, dport=1234) /
62 def create_stream_ip6(self, src_if, mpls_label, mpls_ttl):
64 for i in range(0, 257):
65 info = self.create_packet_info(src_if.sw_if_index,
67 payload = self.info_to_payload(info)
68 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
69 MPLS(label=mpls_label, ttl=mpls_ttl) /
70 IPv6(src=src_if.remote_ip6, dst=src_if.remote_ip6) /
71 UDP(sport=1234, dport=1234) /
77 def verify_capture_ip4(self, src_if, capture, sent):
79 self.assertEqual(len(capture), len(sent))
81 for i in range(len(capture)):
85 # the rx'd packet has the MPLS label popped
87 self.assertEqual(eth.type, 0x800);
92 self.assertEqual(rx_ip.src, tx_ip.src)
93 self.assertEqual(rx_ip.dst, tx_ip.dst)
94 # IP processing post pop has decremented the TTL
95 self.assertEqual(rx_ip.ttl+1, tx_ip.ttl)
100 def verify_capture_ip6(self, src_if, capture, sent):
102 self.assertEqual(len(capture), len(sent))
104 for i in range(len(capture)):
108 # the rx'd packet has the MPLS label popped
110 self.assertEqual(eth.type, 0x86DD);
115 self.assertEqual(rx_ip.src, tx_ip.src)
116 self.assertEqual(rx_ip.dst, tx_ip.dst)
117 # IP processing post pop has decremented the TTL
118 self.assertEqual(rx_ip.hlim + 1, tx_ip.hlim)
124 def test_v4_exp_null(self):
125 """ MPLS V4 Explicit NULL test """
128 # The first test case has an MPLS TTL of 0
129 # all packet should be dropped
131 tx = self.create_stream_ip4(self.pg0, 0, 0)
132 self.pg0.add_stream(tx)
134 self.pg_enable_capture(self.pg_interfaces)
137 rx = self.pg0.get_capture()
140 self.assertEqual(0, len(rx));
142 error("MPLS TTL=0 packets forwarded")
147 # a stream with a non-zero MPLS TTL
148 # PG0 is in the default table
150 self.vapi.cli("clear trace")
151 tx = self.create_stream_ip4(self.pg0, 0, 2)
152 self.pg0.add_stream(tx)
154 self.pg_enable_capture(self.pg_interfaces)
157 rx = self.pg0.get_capture()
158 self.verify_capture_ip4(self.pg0, rx, tx)
161 # a stream with a non-zero MPLS TTL
163 # we are ensuring the post-pop lookup occurs in the VRF table
165 self.vapi.cli("clear trace")
166 tx = self.create_stream_ip4(self.pg1, 0, 2)
167 self.pg1.add_stream(tx)
169 self.pg_enable_capture(self.pg_interfaces)
172 rx = self.pg1.get_capture()
173 self.verify_capture_ip4(self.pg0, rx, tx)
175 def test_v6_exp_null(self):
176 """ MPLS V6 Explicit NULL test """
179 # a stream with a non-zero MPLS TTL
180 # PG0 is in the default table
182 self.vapi.cli("clear trace")
183 tx = self.create_stream_ip6(self.pg0, 2, 2)
184 self.pg0.add_stream(tx)
186 self.pg_enable_capture(self.pg_interfaces)
189 rx = self.pg0.get_capture()
190 self.verify_capture_ip6(self.pg0, rx, tx)
193 # a stream with a non-zero MPLS TTL
195 # we are ensuring the post-pop lookup occurs in the VRF table
197 self.vapi.cli("clear trace")
198 tx = self.create_stream_ip6(self.pg1, 2, 2)
199 self.pg1.add_stream(tx)
201 self.pg_enable_capture(self.pg_interfaces)
204 rx = self.pg1.get_capture()
205 self.verify_capture_ip6(self.pg0, rx, tx)
208 if __name__ == '__main__':
209 unittest.main(testRunner=VppTestRunner)