8800f732f8
- Make framework.py classes a subset of asfframework.py classes - Remove all packet related code from asfframework.py - Add test class and test case set up debug output to log - Repatriate packet tests from asf to test directory - Remove non-packet related code from framework.py and inherit them from asfframework.py classes - Clean up unused import variables - Re-enable BFD tests on Ubuntu 22.04 and fix intermittent test failures in echo_looped_back testcases (where # control packets verified but not guaranteed to be received during test) - Re-enable Wireguard tests on Ubuntu 22.04 and fix intermittent test failures in handshake ratelimiting testcases and event testcase - Run Wiregard testcase suites solo - Improve debug output in log.txt - Increase VCL/LDP post sleep timeout to allow iperf server to finish cleanly. - Fix pcap history files to be sorted by suite and testcase and ensure order/timestamp is correct based on creation in the testcase. - Decode pcap files for each suite and testcase for all errors or if configured via comandline option / env var - Improve vpp corefile detection to allow complete corefile generation - Disable vm vpp interfaces testcases on debian11 - Clean up failed unittest dir when retrying failed testcases and unify testname directory and failed linknames into framwork functions Type: test Change-Id: I0764f79ea5bb639d278bf635ed2408d4d5220e1e Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
314 lines
8.8 KiB
Python
314 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import unittest
|
|
|
|
from framework import VppTestCase
|
|
from asfframework import VppTestRunner
|
|
from vpp_ip_route import (
|
|
VppIpRoute,
|
|
VppRoutePath,
|
|
VppMplsRoute,
|
|
VppMplsTable,
|
|
VppMplsLabel,
|
|
)
|
|
from vpp_mpls_tunnel_interface import VppMPLSTunnelInterface
|
|
|
|
from scapy.packet import Raw
|
|
from scapy.layers.l2 import Ether
|
|
from scapy.layers.inet import IP, UDP
|
|
from scapy.layers.inet6 import IPv6
|
|
from scapy.contrib.mpls import MPLS
|
|
|
|
|
|
def verify_filter(capture, sent):
|
|
if not len(capture) == len(sent):
|
|
# filter out any IPv6 RAs from the capture
|
|
for p in capture:
|
|
if p.haslayer(IPv6):
|
|
capture.remove(p)
|
|
return capture
|
|
|
|
|
|
def verify_mpls_stack(tst, rx, mpls_labels):
|
|
# the rx'd packet has the MPLS label popped
|
|
eth = rx[Ether]
|
|
tst.assertEqual(eth.type, 0x8847)
|
|
|
|
rx_mpls = rx[MPLS]
|
|
|
|
for ii in range(len(mpls_labels)):
|
|
tst.assertEqual(rx_mpls.label, mpls_labels[ii].value)
|
|
tst.assertEqual(rx_mpls.cos, mpls_labels[ii].exp)
|
|
tst.assertEqual(rx_mpls.ttl, mpls_labels[ii].ttl)
|
|
|
|
if ii == len(mpls_labels) - 1:
|
|
tst.assertEqual(rx_mpls.s, 1)
|
|
else:
|
|
# not end of stack
|
|
tst.assertEqual(rx_mpls.s, 0)
|
|
# pop the label to expose the next
|
|
rx_mpls = rx_mpls[MPLS].payload
|
|
|
|
|
|
class TestSRMPLS(VppTestCase):
|
|
"""SR-MPLS Test Case"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TestSRMPLS, cls).setUpClass()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestSRMPLS, cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
super(TestSRMPLS, self).setUp()
|
|
|
|
# create 2 pg interfaces
|
|
self.create_pg_interfaces(range(4))
|
|
|
|
# setup both interfaces
|
|
# assign them different tables.
|
|
table_id = 0
|
|
self.tables = []
|
|
|
|
tbl = VppMplsTable(self, 0)
|
|
tbl.add_vpp_config()
|
|
self.tables.append(tbl)
|
|
|
|
for i in self.pg_interfaces:
|
|
i.admin_up()
|
|
i.config_ip4()
|
|
i.resolve_arp()
|
|
i.config_ip6()
|
|
i.resolve_ndp()
|
|
i.enable_mpls()
|
|
|
|
def tearDown(self):
|
|
for i in self.pg_interfaces:
|
|
i.unconfig_ip4()
|
|
i.unconfig_ip6()
|
|
i.disable_mpls()
|
|
i.admin_down()
|
|
super(TestSRMPLS, self).tearDown()
|
|
|
|
def create_stream_ip4(self, src_if, dst_ip, ip_ttl=64, ip_dscp=0):
|
|
self.reset_packet_infos()
|
|
pkts = []
|
|
for i in range(0, 257):
|
|
info = self.create_packet_info(src_if, src_if)
|
|
payload = self.info_to_payload(info)
|
|
p = (
|
|
Ether(dst=src_if.local_mac, src=src_if.remote_mac)
|
|
/ IP(src=src_if.remote_ip4, dst=dst_ip, ttl=ip_ttl, tos=ip_dscp)
|
|
/ UDP(sport=1234, dport=1234)
|
|
/ Raw(payload)
|
|
)
|
|
info.data = p.copy()
|
|
pkts.append(p)
|
|
return pkts
|
|
|
|
def verify_capture_labelled_ip4(
|
|
self, src_if, capture, sent, mpls_labels, ip_ttl=None
|
|
):
|
|
try:
|
|
capture = verify_filter(capture, sent)
|
|
|
|
self.assertEqual(len(capture), len(sent))
|
|
|
|
for i in range(len(capture)):
|
|
tx = sent[i]
|
|
rx = capture[i]
|
|
tx_ip = tx[IP]
|
|
rx_ip = rx[IP]
|
|
|
|
verify_mpls_stack(self, rx, mpls_labels)
|
|
|
|
self.assertEqual(rx_ip.src, tx_ip.src)
|
|
self.assertEqual(rx_ip.dst, tx_ip.dst)
|
|
if not ip_ttl:
|
|
# IP processing post pop has decremented the TTL
|
|
self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl)
|
|
else:
|
|
self.assertEqual(rx_ip.ttl, ip_ttl)
|
|
|
|
except:
|
|
raise
|
|
|
|
def verify_capture_tunneled_ip4(self, src_if, capture, sent, mpls_labels):
|
|
try:
|
|
capture = verify_filter(capture, sent)
|
|
|
|
self.assertEqual(len(capture), len(sent))
|
|
|
|
for i in range(len(capture)):
|
|
tx = sent[i]
|
|
rx = capture[i]
|
|
tx_ip = tx[IP]
|
|
rx_ip = rx[IP]
|
|
|
|
verify_mpls_stack(self, rx, mpls_labels)
|
|
|
|
self.assertEqual(rx_ip.src, tx_ip.src)
|
|
self.assertEqual(rx_ip.dst, tx_ip.dst)
|
|
# IP processing post pop has decremented the TTL
|
|
self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl)
|
|
|
|
except:
|
|
raise
|
|
|
|
def test_sr_mpls(self):
|
|
"""SR MPLS"""
|
|
|
|
#
|
|
# A simple MPLS xconnect - neos label in label out
|
|
#
|
|
route_32_eos = VppMplsRoute(
|
|
self,
|
|
32,
|
|
0,
|
|
[
|
|
VppRoutePath(
|
|
self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(32)]
|
|
)
|
|
],
|
|
)
|
|
route_32_eos.add_vpp_config()
|
|
|
|
#
|
|
# A binding SID with only one label
|
|
#
|
|
self.vapi.sr_mpls_policy_add(999, 1, 0, [32])
|
|
|
|
#
|
|
# A labeled IP route that resolves thru the binding SID
|
|
#
|
|
ip_10_0_0_1 = VppIpRoute(
|
|
self,
|
|
"10.0.0.1",
|
|
32,
|
|
[
|
|
VppRoutePath(
|
|
"0.0.0.0", 0xFFFFFFFF, nh_via_label=999, labels=[VppMplsLabel(55)]
|
|
)
|
|
],
|
|
)
|
|
ip_10_0_0_1.add_vpp_config()
|
|
|
|
tx = self.create_stream_ip4(self.pg1, "10.0.0.1")
|
|
rx = self.send_and_expect(self.pg1, tx, self.pg0)
|
|
self.verify_capture_labelled_ip4(
|
|
self.pg0, rx, tx, [VppMplsLabel(32), VppMplsLabel(55)]
|
|
)
|
|
|
|
#
|
|
# An unlabeled IP route that resolves thru the binding SID
|
|
#
|
|
ip_10_0_0_1 = VppIpRoute(
|
|
self,
|
|
"10.0.0.2",
|
|
32,
|
|
[VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_via_label=999)],
|
|
)
|
|
ip_10_0_0_1.add_vpp_config()
|
|
|
|
tx = self.create_stream_ip4(self.pg1, "10.0.0.2")
|
|
rx = self.send_and_expect(self.pg1, tx, self.pg0)
|
|
self.verify_capture_labelled_ip4(self.pg0, rx, tx, [VppMplsLabel(32)])
|
|
|
|
self.vapi.sr_mpls_policy_del(999)
|
|
|
|
#
|
|
# this time the SID has many labels pushed
|
|
#
|
|
self.vapi.sr_mpls_policy_add(999, 1, 0, [32, 33, 34])
|
|
|
|
tx = self.create_stream_ip4(self.pg1, "10.0.0.1")
|
|
rx = self.send_and_expect(self.pg1, tx, self.pg0)
|
|
self.verify_capture_labelled_ip4(
|
|
self.pg0,
|
|
rx,
|
|
tx,
|
|
[VppMplsLabel(32), VppMplsLabel(33), VppMplsLabel(34), VppMplsLabel(55)],
|
|
)
|
|
tx = self.create_stream_ip4(self.pg1, "10.0.0.2")
|
|
rx = self.send_and_expect(self.pg1, tx, self.pg0)
|
|
self.verify_capture_labelled_ip4(
|
|
self.pg0, rx, tx, [VppMplsLabel(32), VppMplsLabel(33), VppMplsLabel(34)]
|
|
)
|
|
|
|
#
|
|
# Resolve an MPLS tunnel via the SID
|
|
#
|
|
mpls_tun = VppMPLSTunnelInterface(
|
|
self,
|
|
[
|
|
VppRoutePath(
|
|
"0.0.0.0",
|
|
0xFFFFFFFF,
|
|
nh_via_label=999,
|
|
labels=[VppMplsLabel(44), VppMplsLabel(46)],
|
|
)
|
|
],
|
|
)
|
|
mpls_tun.add_vpp_config()
|
|
mpls_tun.admin_up()
|
|
|
|
#
|
|
# add an unlabelled route through the new tunnel
|
|
#
|
|
route_10_0_0_3 = VppIpRoute(
|
|
self, "10.0.0.3", 32, [VppRoutePath("0.0.0.0", mpls_tun._sw_if_index)]
|
|
)
|
|
route_10_0_0_3.add_vpp_config()
|
|
self.logger.info(self.vapi.cli("sh mpls tun 0"))
|
|
self.logger.info(self.vapi.cli("sh adj 21"))
|
|
|
|
tx = self.create_stream_ip4(self.pg1, "10.0.0.3")
|
|
rx = self.send_and_expect(self.pg1, tx, self.pg0)
|
|
self.verify_capture_tunneled_ip4(
|
|
self.pg0,
|
|
rx,
|
|
tx,
|
|
[
|
|
VppMplsLabel(32),
|
|
VppMplsLabel(33),
|
|
VppMplsLabel(34),
|
|
VppMplsLabel(44),
|
|
VppMplsLabel(46),
|
|
],
|
|
)
|
|
|
|
#
|
|
# add a labelled route through the new tunnel
|
|
#
|
|
route_10_0_0_3 = VppIpRoute(
|
|
self,
|
|
"10.0.0.4",
|
|
32,
|
|
[VppRoutePath("0.0.0.0", mpls_tun._sw_if_index, labels=[VppMplsLabel(55)])],
|
|
)
|
|
route_10_0_0_3.add_vpp_config()
|
|
|
|
tx = self.create_stream_ip4(self.pg1, "10.0.0.4")
|
|
rx = self.send_and_expect(self.pg1, tx, self.pg0)
|
|
self.verify_capture_tunneled_ip4(
|
|
self.pg0,
|
|
rx,
|
|
tx,
|
|
[
|
|
VppMplsLabel(32),
|
|
VppMplsLabel(33),
|
|
VppMplsLabel(34),
|
|
VppMplsLabel(44),
|
|
VppMplsLabel(46),
|
|
VppMplsLabel(55),
|
|
],
|
|
)
|
|
|
|
self.vapi.sr_mpls_policy_del(999)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(testRunner=VppTestRunner)
|