diff options
Diffstat (limited to 'test/test_punt.py')
| -rw-r--r-- | test/test_punt.py | 163 |
1 files changed, 160 insertions, 3 deletions
diff --git a/test/test_punt.py b/test/test_punt.py index 28c17daa08a..598f140547b 100644 --- a/test/test_punt.py +++ b/test/test_punt.py @@ -22,6 +22,7 @@ from scapy.layers.inet import IP, UDP, ICMP from scapy.layers.ipsec import ESP import scapy.layers.inet6 as inet6 from scapy.layers.inet6 import IPv6, ICMPv6DestUnreach +from scapy.contrib.ospf import OSPF_Hdr, OSPFv3_Hello import six from framework import VppTestCase, VppTestRunner @@ -130,6 +131,13 @@ class TestPuntSocket(VppTestCase): self.assertEqual(vpr.punt.punt.exception.id, pr['punt']['exception']['id']) + def verify_ip_proto(self, pr, vpr): + self.assertEqual(vpr.punt.type, pr['type']) + self.assertEqual(vpr.punt.punt.ip_proto.af, + pr['punt']['ip_proto']['af']) + self.assertEqual(vpr.punt.punt.ip_proto.protocol, + pr['punt']['ip_proto']['protocol']) + def verify_udp_pkts(self, rxs, n_rx, port): n_match = 0 for rx in rxs: @@ -182,7 +190,7 @@ def mk_vpp_cfg6(): class TestIP4PuntSocket(TestPuntSocket): - """ Punt Socket for IPv4 """ + """ Punt Socket for IPv4 UDP """ @classmethod def setUpClass(cls): @@ -409,7 +417,7 @@ class TestIP4PuntSocket(TestPuntSocket): class TestIP6PuntSocket(TestPuntSocket): - """ Punt Socket for IPv6""" + """ Punt Socket for IPv6 UDP """ @classmethod def setUpClass(cls): @@ -850,8 +858,157 @@ class TestExceptionPuntSocket(TestPuntSocket): self.vapi.punt_socket_deregister(cfg['vpp']) +class TestIpProtoPuntSocket(TestPuntSocket): + """ Punt Socket for IP packets """ + + @classmethod + def setUpClass(cls): + super(TestIpProtoPuntSocket, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIpProtoPuntSocket, cls).tearDownClass() + + def setUp(self): + super(TestIpProtoPuntSocket, self).setUp() + + for i in self.pg_interfaces: + i.config_ip4() + i.resolve_arp() + + def tearDown(self): + super(TestIpProtoPuntSocket, self).tearDown() + for i in self.pg_interfaces: + i.unconfig_ip4() + i.admin_down() + + def test_registration(self): + """ Punt socket registration/deregistration""" + + af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4 + pt_ip = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_IP_PROTO + proto_ospf = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_OSPF + proto_eigrp = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_EIGRP + + punts = self.vapi.punt_socket_dump(type=pt_ip) + self.assertEqual(len(punts), 0) + + # + # configure a punt socket + # + punt_ospf = { + 'type': pt_ip, + 'punt': { + 'ip_proto': { + 'af': af_ip4, + 'protocol': proto_ospf + } + } + } + punt_eigrp = { + 'type': pt_ip, + 'punt': { + 'ip_proto': { + 'af': af_ip4, + 'protocol': proto_eigrp + } + } + } + + self.vapi.punt_socket_register(punt_ospf, + b"%s/socket_punt_1" % + six.ensure_binary(self.tempdir)) + self.vapi.punt_socket_register(punt_eigrp, + b"%s/socket_punt_2" % + six.ensure_binary(self.tempdir)) + self.logger.info(self.vapi.cli("sh punt sock reg ip")) + punts = self.vapi.punt_socket_dump(type=pt_ip) + self.assertEqual(len(punts), 2) + self.verify_ip_proto(punt_ospf, punts[0]) + self.verify_ip_proto(punt_eigrp, punts[1]) + + # + # deregister a punt socket + # + self.vapi.punt_socket_deregister(punt_ospf) + punts = self.vapi.punt_socket_dump(type=pt_ip) + self.assertEqual(len(punts), 1) + + # + # configure a punt socket again + # + self.vapi.punt_socket_register(punt_ospf, + b"%s/socket_punt_3" % + six.ensure_binary(self.tempdir)) + punts = self.vapi.punt_socket_dump(type=pt_ip) + self.assertEqual(len(punts), 2) + + self.logger.info(self.vapi.cli("sh punt sock reg exception")) + + # + # deregister all punt socket + # + self.vapi.punt_socket_deregister(punt_eigrp) + self.vapi.punt_socket_deregister(punt_ospf) + punts = self.vapi.punt_socket_dump(type=pt_ip) + self.assertEqual(len(punts), 0) + + def verify_ospf_pkts(self, rxs, n_sent): + self.assertEqual(len(rxs), n_sent) + for rx in rxs: + self.assertTrue(rx.haslayer(OSPF_Hdr)) + + def test_traffic(self): + """ Punt socket traffic """ + + af_ip4 = VppEnum.vl_api_address_family_t.ADDRESS_IP4 + pt_ip = VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_IP_PROTO + proto_ospf = VppEnum.vl_api_ip_proto_t.IP_API_PROTO_OSPF + + # + # configure a punt socket to capture OSPF packets + # + punt_ospf = { + 'type': pt_ip, + 'punt': { + 'ip_proto': { + 'af': af_ip4, + 'protocol': proto_ospf + } + } + } + + # + # create packet streams and configure a punt sockets + # + pkt = (Ether(src=self.pg0.remote_mac, + dst=self.pg0.local_mac) / + IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / + OSPF_Hdr() / + OSPFv3_Hello()) + pkts = pkt * 7 + + sock = self.socket_client_create(b"%s/socket_1" % ( + six.ensure_binary(self.tempdir))) + self.vapi.punt_socket_register( + punt_ospf, + b"%s/socket_1" % (six.ensure_binary(self.tempdir))) + + # + # send packets for each SPI we expect to be punted + # + self.send_and_assert_no_replies(self.pg0, pkts) + + # + # verify the punted packets arrived on the associated socket + # + rx = sock.close() + self.verify_ospf_pkts(rx, len(pkts)) + self.vapi.punt_socket_deregister(punt_ospf) + + class TestPunt(VppTestCase): - """ Punt Test Case """ + """ Exception Punt Test Case """ @classmethod def setUpClass(cls): |