from framework import VppTestCase, VppTestRunner
from vpp_udp_encap import *
-from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable
+from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, VppMplsLabel
from scapy.packet import Raw
from scapy.layers.l2 import Ether, ARP
self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
def validate_inner4(self, rx, tx, ttl=None):
- self.assertEqual(rx.src, tx[IP].src)
- self.assertEqual(rx.dst, tx[IP].dst)
+ self.assertEqual(rx[IP].src, tx[IP].src)
+ self.assertEqual(rx[IP].dst, tx[IP].dst)
if ttl:
- self.assertEqual(rx.ttl, ttl)
+ self.assertEqual(rx[IP].ttl, ttl)
else:
- self.assertEqual(rx.ttl, tx[IP].ttl)
+ self.assertEqual(rx[IP].ttl, tx[IP].ttl)
def validate_inner6(self, rx, tx):
self.assertEqual(rx.src, tx[IPv6].src)
self.assertEqual(rx.dst, tx[IPv6].dst)
self.assertEqual(rx.hlim, tx[IPv6].hlim)
- def send_and_expect(self, input, output, pkts):
- self.vapi.cli("clear trace")
- input.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- rx = output.get_capture(len(pkts))
- return rx
-
def test_udp_encap(self):
""" UDP Encap test
"""
IP(src="2.2.2.2", dst="1.1.0.1") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, self.pg0, p_4o4*65)
+ rx = self.send_and_expect(self.pg0, p_4o4*65, self.pg0)
for p in rx:
self.validate_outer4(p, udp_encap_0)
p = IP(p["UDP"].payload.load)
IP(src="2.2.2.2", dst="1.1.2.1") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, self.pg2, p_4o6*65)
+ rx = self.send_and_expect(self.pg0, p_4o6*65, self.pg2)
for p in rx:
self.validate_outer6(p, udp_encap_2)
p = IP(p["UDP"].payload.load)
IPv6(src="2001::100", dst="2001::1") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, self.pg1, p_6o4*65)
+ rx = self.send_and_expect(self.pg0, p_6o4*65, self.pg1)
for p in rx:
self.validate_outer4(p, udp_encap_1)
p = IPv6(p["UDP"].payload.load)
IPv6(src="2001::100", dst="2001::3") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, self.pg3, p_6o6*65)
+ rx = self.send_and_expect(self.pg0, p_6o6*65, self.pg3)
for p in rx:
self.validate_outer6(p, udp_encap_3)
p = IPv6(p["UDP"].payload.load)
0xFFFFFFFF,
is_udp_encap=1,
next_hop_id=1,
- labels=[66])])
+ labels=[VppMplsLabel(66)])])
route_4oMPLSo4.add_vpp_config()
p_4omo4 = (Ether(src=self.pg0.remote_mac,
IP(src="2.2.2.2", dst="1.1.2.22") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, self.pg1, p_4omo4*65)
+ rx = self.send_and_expect(self.pg0, p_4omo4*65, self.pg1)
for p in rx:
self.validate_outer4(p, udp_encap_1)
p = MPLS(p["UDP"].payload.load)