diff options
Diffstat (limited to 'test')
| -rwxr-xr-x | test/framework.py | 30 | ||||
| -rw-r--r-- | test/test_pg.py | 98 | ||||
| -rwxr-xr-x | test/vpp_pg_interface.py | 5 |
3 files changed, 129 insertions, 4 deletions
diff --git a/test/framework.py b/test/framework.py index c2a2fc68430..2afec7a4f0f 100755 --- a/test/framework.py +++ b/test/framework.py @@ -32,6 +32,7 @@ from vpp_sub_interface import VppSubInterface from vpp_lo_interface import VppLoInterface from vpp_bvi_interface import VppBviInterface from vpp_papi_provider import VppPapiProvider +from vpp_papi import VppEnum import vpp_papi from vpp_papi.vpp_stats import VPPStats from vpp_papi.vpp_transport_socket import VppTransportSocketIOError @@ -915,7 +916,8 @@ class VppTestCase(CPUInterface, unittest.TestCase): cls._pcaps = [] @classmethod - def create_pg_interfaces(cls, interfaces, gso=0, gso_size=0): + def create_pg_interfaces_internal(cls, interfaces, gso=0, gso_size=0, + mode=None): """ Create packet-generator interfaces. @@ -925,13 +927,37 @@ class VppTestCase(CPUInterface, unittest.TestCase): """ result = [] for i in interfaces: - intf = VppPGInterface(cls, i, gso, gso_size) + intf = VppPGInterface(cls, i, gso, gso_size, mode) setattr(cls, intf.name, intf) result.append(intf) cls.pg_interfaces = result return result @classmethod + def create_pg_ip4_interfaces(cls, interfaces, gso=0, gso_size=0): + pgmode = VppEnum.vl_api_pg_interface_mode_t + return cls.create_pg_interfaces_internal(interfaces, gso, gso_size, + pgmode.PG_API_MODE_IP4) + + @classmethod + def create_pg_ip6_interfaces(cls, interfaces, gso=0, gso_size=0): + pgmode = VppEnum.vl_api_pg_interface_mode_t + return cls.create_pg_interfaces_internal(interfaces, gso, gso_size, + pgmode.PG_API_MODE_IP6) + + @classmethod + def create_pg_interfaces(cls, interfaces, gso=0, gso_size=0): + pgmode = VppEnum.vl_api_pg_interface_mode_t + return cls.create_pg_interfaces_internal(interfaces, gso, gso_size, + pgmode.PG_API_MODE_ETHERNET) + + @classmethod + def create_pg_ethernet_interfaces(cls, interfaces, gso=0, gso_size=0): + pgmode = VppEnum.vl_api_pg_interface_mode_t + return cls.create_pg_interfaces_internal(interfaces, gso, gso_size, + pgmode.PG_API_MODE_ETHERNET) + + @classmethod def create_loopback_interfaces(cls, count): """ Create loopback interfaces. diff --git a/test/test_pg.py b/test/test_pg.py new file mode 100644 index 00000000000..76b7fd7f1d9 --- /dev/null +++ b/test/test_pg.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 + +import unittest + +import scapy.compat +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 framework import VppTestCase, VppTestRunner + + +class TestPgTun(VppTestCase): + """ PG Test Case """ + + def setUp(self): + super(TestPgTun, self).setUp() + + # create 3 pg interfaces - one each ethernet, ip4-tun, ip6-tun. + self.create_pg_interfaces(range(0, 1)) + self.pg_interfaces += self.create_pg_ip4_interfaces(range(1, 2)) + self.pg_interfaces += self.create_pg_ip6_interfaces(range(2, 3)) + + for i in self.pg_interfaces: + i.admin_up() + + for i in [self.pg0, self.pg1]: + i.config_ip4() + + for i in [self.pg0, self.pg2]: + i.config_ip6() + + self.pg0.resolve_arp() + self.pg0.resolve_ndp() + + def tearDown(self): + for i in self.pg_interfaces: + i.unconfig_ip4() + i.admin_down() + super(TestPgTun, self).tearDown() + + def test_pg_tun(self): + """ IP[46] Tunnel Mode PG """ + + # + # test that we can send and receive IP encap'd packets on the + # tun interfaces + # + N_PKTS = 31 + + # v4 tun to ethernet + p = (IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) / + UDP(sport=1234, dport=1234) / + Raw('0' * 48)) + + rxs = self.send_and_expect(self.pg1, p * N_PKTS, self.pg0) + for rx in rxs: + self.assertEqual(rx[Ether].dst, self.pg0.remote_mac) + self.assertEqual(rx[IP].dst, self.pg0.remote_ip4) + + # v6 tun to ethernet + p = (IPv6(src=self.pg2.remote_ip6, dst=self.pg0.remote_ip6) / + UDP(sport=1234, dport=1234) / + Raw('0' * 48)) + + rxs = self.send_and_expect(self.pg2, p * N_PKTS, self.pg0) + for rx in rxs: + self.assertEqual(rx[Ether].dst, self.pg0.remote_mac) + self.assertEqual(rx[IPv6].dst, self.pg0.remote_ip6) + + # eth to v4 tun + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) / + UDP(sport=1234, dport=1234) / + Raw('0' * 48)) + + rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg1) + for rx in rxs: + rx = IP(rx) + self.assertFalse(rx.haslayer(Ether)) + self.assertEqual(rx[IP].dst, self.pg1.remote_ip4) + + # eth to v6 tun + p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / + IPv6(src=self.pg0.remote_ip6, dst=self.pg2.remote_ip6) / + UDP(sport=1234, dport=1234) / + Raw('0' * 48)) + + rxs = self.send_and_expect(self.pg0, p * N_PKTS, self.pg2) + for rx in rxs: + rx = IPv6(rx) + self.assertFalse(rx.haslayer(Ether)) + self.assertEqual(rx[IPv6].dst, self.pg2.remote_ip6) + + +if __name__ == '__main__': + unittest.main(testRunner=VppTestRunner) diff --git a/test/vpp_pg_interface.py b/test/vpp_pg_interface.py index 3bcde9319a8..567b81de7c0 100755 --- a/test/vpp_pg_interface.py +++ b/test/vpp_pg_interface.py @@ -9,6 +9,7 @@ import scapy.compat from scapy.utils import wrpcap, rdpcap, PcapReader from scapy.plist import PacketList from vpp_interface import VppInterface +from vpp_papi import VppEnum from scapy.layers.l2 import Ether, ARP from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA,\ @@ -111,11 +112,11 @@ class VppPGInterface(VppInterface): self._out_history_counter += 1 return v - def __init__(self, test, pg_index, gso, gso_size): + def __init__(self, test, pg_index, gso, gso_size, mode): """ Create VPP packet-generator interface """ super().__init__(test) - r = test.vapi.pg_create_interface(pg_index, gso, gso_size) + r = test.vapi.pg_create_interface_v2(pg_index, gso, gso_size, mode) self.set_sw_if_index(r.sw_if_index) self._in_history_counter = 0 |