X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_punt.py;h=2a245385d851e1b2eb5a8ba76e9c9662791388d1;hb=22ab6f7cbb0f6139302aa6ca9f0c96dba17a37a7;hp=b20dc76c61d3e0825d0fea11c00093a0e42a31ea;hpb=919efad2671993d4c6d5a0dba8eeb99d5c60edf1;p=vpp.git diff --git a/test/test_punt.py b/test/test_punt.py index b20dc76c61d..2a245385d85 100644 --- a/test/test_punt.py +++ b/test/test_punt.py @@ -2,30 +2,36 @@ import binascii import random import socket -import unittest import os -import scapy.layers.inet6 as inet6 import threading import struct - from struct import unpack, unpack_from + +try: + import unittest2 as unittest +except ImportError: + import unittest + from util import ppp, ppc from re import compile +import scapy.compat from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP, ICMP +import scapy.layers.inet6 as inet6 from scapy.layers.inet6 import IPv6, ICMPv6DestUnreach +import six from framework import VppTestCase, VppTestRunner # Format MAC Address def get_mac_addr(bytes_addr): - return ':'.join('%02x' % ord(b) for b in bytes_addr) + return ':'.join('%02x' % scapy.compat.orb(b) for b in bytes_addr) # Format IP Address def ipv4(bytes_addr): - return '.'.join('%d' % ord(b) for b in bytes_addr) + return '.'.join('%d' % scapy.compat.orb(b) for b in bytes_addr) # Unpack Ethernet Frame @@ -119,10 +125,18 @@ class TestPuntSocket(VppTestCase): portsCheck = dict() nr_packets = 256 + @classmethod + def setUpClass(cls): + super(TestPuntSocket, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestPuntSocket, cls).tearDownClass() + @classmethod def setUpConstants(cls): - tempdir = cls.tempdir - cls.config.add('punt', 'socket', '%s/socket_punt' % cls.tempdir) + cls.extra_vpp_punt_config = [ + "punt", "{", "socket", cls.tempdir+"/socket_punt", "}"] super(TestPuntSocket, cls).setUpConstants() def setUp(self): @@ -135,6 +149,7 @@ class TestPuntSocket(VppTestCase): def tearDown(self): del self.sock_servers[:] + super(TestPuntSocket, self).tearDown() def socket_client_create(self, sock_name, id=None): thread = serverSocketThread(id, sock_name, self.portsCheck) @@ -149,6 +164,14 @@ class TestPuntSocket(VppTestCase): class TestIP4PuntSocket(TestPuntSocket): """ Punt Socket for IPv4 """ + @classmethod + def setUpClass(cls): + super(TestIP4PuntSocket, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP4PuntSocket, cls).tearDownClass() + def setUp(self): super(TestIP4PuntSocket, self).setUp() @@ -171,8 +194,10 @@ class TestIP4PuntSocket(TestPuntSocket): # # configure a punt socket # - self.vapi.punt_socket_register(1111, self.tempdir+"/socket_punt_1111") - self.vapi.punt_socket_register(2222, self.tempdir+"/socket_punt_2222") + self.vapi.punt_socket_register(1111, b"%s/socket_punt_1111" % + six.ensure_binary(self.tempdir)) + self.vapi.punt_socket_register(2222, b"%s/socket_punt_2222" % + six.ensure_binary(self.tempdir)) punts = self.vapi.punt_socket_dump(is_ip6=0) self.assertEqual(len(punts), 2) self.assertEqual(punts[0].punt.l4_port, 1111) @@ -188,8 +213,10 @@ class TestIP4PuntSocket(TestPuntSocket): # # configure a punt socket again # - self.vapi.punt_socket_register(1111, self.tempdir+"/socket_punt_1111") - self.vapi.punt_socket_register(3333, self.tempdir+"/socket_punt_3333") + self.vapi.punt_socket_register(1111, b"%s/socket_punt_1111" % + six.ensure_binary(self.tempdir)) + self.vapi.punt_socket_register(3333, b"%s/socket_punt_3333" % + six.ensure_binary(self.tempdir)) punts = self.vapi.punt_socket_dump(is_ip6=0) self.assertEqual(len(punts), 3) @@ -235,9 +262,10 @@ class TestIP4PuntSocket(TestPuntSocket): # # configure a punt socket # - self.socket_client_create(self.tempdir+"/socket_" + str(port)) - self.vapi.punt_socket_register(port, self.tempdir+"/socket_" + - str(port)) + self.socket_client_create(b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), port)) + self.vapi.punt_socket_register(port, b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), port)) punts = self.vapi.punt_socket_dump(is_ip6=0) self.assertEqual(len(punts), 1) @@ -298,8 +326,10 @@ class TestIP4PuntSocket(TestPuntSocket): # configure a punt socket # for p in self.ports: - self.socket_client_create(self.tempdir+"/socket_" + str(p)) - self.vapi.punt_socket_register(p, self.tempdir+"/socket_" + str(p)) + self.socket_client_create(b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), p)) + self.vapi.punt_socket_register(p, b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), p)) punts = self.vapi.punt_socket_dump(is_ip6=0) self.assertEqual(len(punts), len(self.ports)) @@ -354,9 +384,12 @@ class TestIP4PuntSocket(TestPuntSocket): # configure a punt socket # - self.socket_client_create(self.tempdir+"/socket_multi") + self.socket_client_create(b"%s/socket_multi" % + six.ensure_binary(self.tempdir)) for p in self.ports: - self.vapi.punt_socket_register(p, self.tempdir+"/socket_multi") + self.vapi.punt_socket_register(p, + b"%s/socket_multi" % + six.ensure_binary(self.tempdir)) punts = self.vapi.punt_socket_dump(is_ip6=0) self.assertEqual(len(punts), len(self.ports)) @@ -385,6 +418,14 @@ class TestIP4PuntSocket(TestPuntSocket): class TestIP6PuntSocket(TestPuntSocket): """ Punt Socket for IPv6""" + @classmethod + def setUpClass(cls): + super(TestIP6PuntSocket, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestIP6PuntSocket, cls).tearDownClass() + def setUp(self): super(TestIP6PuntSocket, self).setUp() @@ -407,9 +448,11 @@ class TestIP6PuntSocket(TestPuntSocket): # # configure a punt socket # - self.vapi.punt_socket_register(1111, self.tempdir+"/socket_1111", + self.vapi.punt_socket_register(1111, b"%s/socket_1111" % + six.ensure_binary(self.tempdir), is_ip4=0) - self.vapi.punt_socket_register(2222, self.tempdir+"/socket_2222", + self.vapi.punt_socket_register(2222, b"%s/socket_2222" % + six.ensure_binary(self.tempdir), is_ip4=0) punts = self.vapi.punt_socket_dump(is_ip6=1) self.assertEqual(len(punts), 2) @@ -426,7 +469,8 @@ class TestIP6PuntSocket(TestPuntSocket): # # configure a punt socket again # - self.vapi.punt_socket_register(1111, self.tempdir+"/socket_1111", + self.vapi.punt_socket_register(1111, b"%s/socket_1111" % + six.ensure_binary(self.tempdir), is_ip4=0) punts = self.vapi.punt_socket_dump(is_ip6=1) self.assertEqual(len(punts), 2) @@ -473,9 +517,10 @@ class TestIP6PuntSocket(TestPuntSocket): # # configure a punt socket # - self.socket_client_create(self.tempdir+"/socket_" + str(port)) - self.vapi.punt_socket_register(port, self.tempdir+"/socket_" + - str(port), is_ip4=0) + self.socket_client_create(b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), port)) + self.vapi.punt_socket_register(port, b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), port), is_ip4=0) punts = self.vapi.punt_socket_dump(is_ip6=1) self.assertEqual(len(punts), 1) @@ -536,9 +581,10 @@ class TestIP6PuntSocket(TestPuntSocket): # configure a punt socket # for p in self.ports: - self.socket_client_create(self.tempdir+"/socket_" + str(p)) - self.vapi.punt_socket_register(p, self.tempdir+"/socket_" + str(p), - is_ip4=0) + self.socket_client_create(b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), p)) + self.vapi.punt_socket_register(p, b"%s/socket_%d" % ( + six.ensure_binary(self.tempdir), p), is_ip4=0) punts = self.vapi.punt_socket_dump(is_ip6=1) self.assertEqual(len(punts), len(self.ports)) @@ -594,9 +640,12 @@ class TestIP6PuntSocket(TestPuntSocket): # # configure a punt socket # - self.socket_client_create(self.tempdir+"/socket_multi") + self.socket_client_create(b"%s/socket_multi" % + six.ensure_binary(self.tempdir)) for p in self.ports: - self.vapi.punt_socket_register(p, self.tempdir+"/socket_multi", + self.vapi.punt_socket_register(p, + b"%s/socket_multi" % + six.ensure_binary(self.tempdir), is_ip4=0) punts = self.vapi.punt_socket_dump(is_ip6=1) self.assertEqual(len(punts), len(self.ports))