ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
vpp/test/test_dslite.py
Dave Wallace 8800f732f8 tests: refactor asf framework code 
- 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>
2023-11-03 05:06:43 +00:00

344 lines
13 KiB
Python
Raw Blame History

#!/usr/bin/env python3
import socket
from asfframework import tag_fixme_vpp_workers
from framework import VppTestCase
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet6 import (
IPv6,
ICMPv6EchoRequest,
ICMPv6EchoReply,
)
from scapy.layers.l2 import Ether
from scapy.data import IP_PROTOS
from scapy.packet import Raw
from syslog_rfc5424_parser import SyslogMessage, ParseError
from syslog_rfc5424_parser.constants import SyslogSeverity
from vpp_ip_route import VppIpRoute, VppRoutePath
@tag_fixme_vpp_workers
class TestDSlite(VppTestCase):
"""DS-Lite Test Cases"""
@classmethod
def setUpClass(cls):
super(TestDSlite, cls).setUpClass()
try:
cls.nat_addr = "10.0.0.3"
cls.create_pg_interfaces(range(3))
cls.pg0.admin_up()
cls.pg0.config_ip4()
cls.pg0.resolve_arp()
cls.pg1.admin_up()
cls.pg1.config_ip6()
cls.pg1.generate_remote_hosts(2)
cls.pg1.configure_ipv6_neighbors()
cls.pg2.admin_up()
cls.pg2.config_ip4()
cls.pg2.resolve_arp()
except Exception:
super(TestDSlite, cls).tearDownClass()
raise
@classmethod
def tearDownClass(cls):
super(TestDSlite, cls).tearDownClass()
def verify_syslog_apmadd(self, data, isaddr, isport, xsaddr, xsport, sv6enc, proto):
message = data.decode("utf-8")
try:
message = SyslogMessage.parse(message)
except ParseError as e:
self.logger.error(e)
else:
self.assertEqual(message.severity, SyslogSeverity.info)
self.assertEqual(message.appname, "NAT")
self.assertEqual(message.msgid, "APMADD")
sd_params = message.sd.get("napmap")
self.assertTrue(sd_params is not None)
self.assertEqual(sd_params.get("IATYP"), "IPv4")
self.assertEqual(sd_params.get("ISADDR"), isaddr)
self.assertEqual(sd_params.get("ISPORT"), "%d" % isport)
self.assertEqual(sd_params.get("XATYP"), "IPv4")
self.assertEqual(sd_params.get("XSADDR"), xsaddr)
self.assertEqual(sd_params.get("XSPORT"), "%d" % xsport)
self.assertEqual(sd_params.get("PROTO"), "%d" % proto)
self.assertTrue(sd_params.get("SSUBIX") is not None)
self.assertEqual(sd_params.get("SV6ENC"), sv6enc)
def test_dslite(self):
"""Test DS-Lite"""
self.vapi.dslite_add_del_pool_addr_range(
start_addr=self.nat_addr, end_addr=self.nat_addr, is_add=1
)
aftr_ip4 = "192.0.0.1"
aftr_ip6 = "2001:db8:85a3::8a2e:370:1"
self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
self.vapi.syslog_set_sender(self.pg2.local_ip4, self.pg2.remote_ip4)
# UDP
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6)
/ IP(dst=self.pg0.remote_ip4, src="192.168.1.1")
/ UDP(sport=20000, dport=10000)
)
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg0.get_capture(1)
capture = capture[0]
self.assertFalse(capture.haslayer(IPv6))
self.assertEqual(capture[IP].src, self.nat_addr)
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertNotEqual(capture[UDP].sport, 20000)
self.assertEqual(capture[UDP].dport, 10000)
self.assert_packet_checksums_valid(capture)
out_port = capture[UDP].sport
capture = self.pg2.get_capture(1)
self.verify_syslog_apmadd(
capture[0][Raw].load,
"192.168.1.1",
20000,
self.nat_addr,
out_port,
self.pg1.remote_hosts[0].ip6,
IP_PROTOS.udp,
)
p = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(dst=self.nat_addr, src=self.pg0.remote_ip4)
/ UDP(sport=10000, dport=out_port)
)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
capture = capture[0]
self.assertEqual(capture[IPv6].src, aftr_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
self.assertEqual(capture[IP].dst, "192.168.1.1")
self.assertEqual(capture[UDP].sport, 10000)
self.assertEqual(capture[UDP].dport, 20000)
self.assert_packet_checksums_valid(capture)
# TCP
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6)
/ IP(dst=self.pg0.remote_ip4, src="192.168.1.1")
/ TCP(sport=20001, dport=10001)
)
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg0.get_capture(1)
capture = capture[0]
self.assertFalse(capture.haslayer(IPv6))
self.assertEqual(capture[IP].src, self.nat_addr)
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertNotEqual(capture[TCP].sport, 20001)
self.assertEqual(capture[TCP].dport, 10001)
self.assert_packet_checksums_valid(capture)
out_port = capture[TCP].sport
p = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(dst=self.nat_addr, src=self.pg0.remote_ip4)
/ TCP(sport=10001, dport=out_port)
)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
capture = capture[0]
self.assertEqual(capture[IPv6].src, aftr_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
self.assertEqual(capture[IP].dst, "192.168.1.1")
self.assertEqual(capture[TCP].sport, 10001)
self.assertEqual(capture[TCP].dport, 20001)
self.assert_packet_checksums_valid(capture)
# ICMP
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6)
/ IP(dst=self.pg0.remote_ip4, src="192.168.1.1")
/ ICMP(id=4000, type="echo-request")
)
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg0.get_capture(1)
capture = capture[0]
self.assertFalse(capture.haslayer(IPv6))
self.assertEqual(capture[IP].src, self.nat_addr)
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertNotEqual(capture[ICMP].id, 4000)
self.assert_packet_checksums_valid(capture)
out_id = capture[ICMP].id
p = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(dst=self.nat_addr, src=self.pg0.remote_ip4)
/ ICMP(id=out_id, type="echo-reply")
)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
capture = capture[0]
self.assertEqual(capture[IPv6].src, aftr_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
self.assertEqual(capture[IP].dst, "192.168.1.1")
self.assertEqual(capture[ICMP].id, 4000)
self.assert_packet_checksums_valid(capture)
# ping DS-Lite AFTR tunnel endpoint address
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(src=self.pg1.remote_hosts[1].ip6, dst=aftr_ip6)
/ ICMPv6EchoRequest()
)
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
capture = capture[0]
self.assertEqual(capture[IPv6].src, aftr_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
self.assertTrue(capture.haslayer(ICMPv6EchoReply))
b4s = self.statistics.get_counter("/dslite/total-b4s")
self.assertEqual(b4s[0][0], 2)
sessions = self.statistics.get_counter("/dslite/total-sessions")
self.assertEqual(sessions[0][0], 3)
def tearDown(self):
super(TestDSlite, self).tearDown()
def show_commands_at_teardown(self):
self.logger.info(self.vapi.cli("show dslite pool"))
self.logger.info(self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
self.logger.info(self.vapi.cli("show dslite sessions"))
class TestDSliteCE(VppTestCase):
"""DS-Lite CE Test Cases"""
@classmethod
def setUpConstants(cls):
super(TestDSliteCE, cls).setUpConstants()
cls.vpp_cmdline.extend(["dslite", "{", "ce", "}"])
@classmethod
def setUpClass(cls):
super(TestDSliteCE, cls).setUpClass()
try:
cls.create_pg_interfaces(range(2))
cls.pg0.admin_up()
cls.pg0.config_ip4()
cls.pg0.resolve_arp()
cls.pg1.admin_up()
cls.pg1.config_ip6()
cls.pg1.generate_remote_hosts(1)
cls.pg1.configure_ipv6_neighbors()
except Exception:
super(TestDSliteCE, cls).tearDownClass()
raise
@classmethod
def tearDownClass(cls):
super(TestDSliteCE, cls).tearDownClass()
def test_dslite_ce(self):
"""Test DS-Lite CE"""
b4_ip4 = "192.0.0.2"
b4_ip6 = "2001:db8:62aa::375e:f4c1:1"
self.vapi.dslite_set_b4_addr(ip4_addr=b4_ip4, ip6_addr=b4_ip6)
aftr_ip4 = "192.0.0.1"
aftr_ip6 = "2001:db8:85a3::8a2e:370:1"
aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
r1 = VppIpRoute(
self,
aftr_ip6,
128,
[VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
)
r1.add_vpp_config()
# UDP encapsulation
p = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4)
/ UDP(sport=10000, dport=20000)
)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
capture = capture[0]
self.assertEqual(capture[IPv6].src, b4_ip6)
self.assertEqual(capture[IPv6].dst, aftr_ip6)
self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
self.assertEqual(capture[UDP].sport, 10000)
self.assertEqual(capture[UDP].dport, 20000)
self.assert_packet_checksums_valid(capture)
# UDP decapsulation
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(dst=b4_ip6, src=aftr_ip6)
/ IP(dst=self.pg0.remote_ip4, src=self.pg1.remote_ip4)
/ UDP(sport=20000, dport=10000)
)
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg0.get_capture(1)
capture = capture[0]
self.assertFalse(capture.haslayer(IPv6))
self.assertEqual(capture[IP].src, self.pg1.remote_ip4)
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertEqual(capture[UDP].sport, 20000)
self.assertEqual(capture[UDP].dport, 10000)
self.assert_packet_checksums_valid(capture)
# ping DS-Lite B4 tunnel endpoint address
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(src=self.pg1.remote_hosts[0].ip6, dst=b4_ip6)
/ ICMPv6EchoRequest()
)
self.pg1.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(1)
capture = capture[0]
self.assertEqual(capture[IPv6].src, b4_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
self.assertTrue(capture.haslayer(ICMPv6EchoReply))
def tearDown(self):
super(TestDSliteCE, self).tearDown()
def show_commands_at_teardown(self):
self.logger.info(self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
self.logger.info(self.vapi.cli("show dslite b4-tunnel-endpoint-address"))
Reference in New Issue View Git Blame Copy Permalink