PAPI: Add MACAddress object wrapper for vl_api_mac_address_t

[vpp.git] / test / util.py

diff --git a/test/util.py b/test/util.py
index 512bf9e..9652b80 100644 (file)
--- a/test/util.py
+++ b/test/util.py
@@ -2,24 +2,29 @@
 
 import socket
 import sys
 
 import socket
 import sys

+import os.path

 from abc import abstractmethod, ABCMeta
 from abc import abstractmethod, ABCMeta

-from cStringIO import StringIO
-from scapy.layers.inet6 import in6_mactoifaceid
+from scapy.utils6 import in6_mactoifaceid

 
 from scapy.layers.l2 import Ether
 from scapy.packet import Raw
 from scapy.layers.inet import IP
 from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
     IPv6ExtHdrHopByHop
 
 from scapy.layers.l2 import Ether
 from scapy.packet import Raw
 from scapy.layers.inet import IP
 from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
     IPv6ExtHdrHopByHop

+from scapy.utils import hexdump

 from socket import AF_INET6
 from socket import AF_INET6

+from io import BytesIO
+from vpp_papi import mac_pton

 
 
 def ppp(headline, packet):
     """ Return string containing the output of scapy packet.show() call. """
 
 
 def ppp(headline, packet):
     """ Return string containing the output of scapy packet.show() call. """

-    o = StringIO()
+    o = BytesIO()

     old_stdout = sys.stdout
     sys.stdout = o
     print(headline)
     old_stdout = sys.stdout
     sys.stdout = o
     print(headline)

+    hexdump(packet)
+    print("")

     packet.show()
     sys.stdout = old_stdout
     return o.getvalue()
     packet.show()
     sys.stdout = old_stdout
     return o.getvalue()


@@ -54,17 +59,45 @@ def ip4n_range(ip4n, s, e):
             for ip in ip4_range(ip4, s, e))
 
 
             for ip in ip4_range(ip4, s, e))
 
 

-def mactobinary(mac):
-    """ Convert the : separated format into binary packet data for the API """
-    return mac.replace(':', '').decode('hex')
-
-

 def mk_ll_addr(mac):
     euid = in6_mactoifaceid(mac)
     addr = "fe80::" + euid
     return addr
 
 
 def mk_ll_addr(mac):
     euid = in6_mactoifaceid(mac)
     addr = "fe80::" + euid
     return addr
 
 

+def ip6_normalize(ip6):
+    return socket.inet_ntop(socket.AF_INET6,
+                            socket.inet_pton(socket.AF_INET6, ip6))
+
+
+def get_core_path(tempdir):
+    return "%s/%s" % (tempdir, get_core_pattern())
+
+
+def is_core_present(tempdir):
+    return os.path.isfile(get_core_path(tempdir))
+
+
+def get_core_pattern():
+    with open("/proc/sys/kernel/core_pattern", "r") as f:
+        corefmt = f.read().strip()
+    return corefmt
+
+
+def check_core_path(logger, core_path):
+    corefmt = get_core_pattern()
+    if corefmt.startswith("|"):
+        logger.error(
+            "WARNING: redirecting the core dump through a"
+            " filter may result in truncated dumps.")
+        logger.error(
+            "   You may want to check the filter settings"
+            " or uninstall it and edit the"
+            " /proc/sys/kernel/core_pattern accordingly.")
+        logger.error(
+            "   current core pattern is: %s" % corefmt)
+
+

 class NumericConstant(object):
     __metaclass__ = ABCMeta
 
 class NumericConstant(object):
     __metaclass__ = ABCMeta
 


@@ -97,7 +130,7 @@ class Host(object):
     @property
     def bin_mac(self):
         """ MAC address """
     @property
     def bin_mac(self):
         """ MAC address """

-        return mactobinary(self._mac)
+        return mac_pton(self._mac)

 
     @property
     def ip4(self):
 
     @property
     def ip4(self):


@@ -404,3 +437,29 @@ def fragment_rfc8200(packet, identification, fragsize, _logger=None):
     pkts[-1][IPv6ExtHdrFragment].m = 0  # reset more-flags in last fragment
 
     return pkts
     pkts[-1][IPv6ExtHdrFragment].m = 0  # reset more-flags in last fragment
 
     return pkts

+
+
+def reassemble4_core(listoffragments, return_ip):
+    buffer = BytesIO()
+    first = listoffragments[0]
+    buffer.seek(20)
+    for pkt in listoffragments:
+        buffer.seek(pkt[IP].frag*8)
+        buffer.write(bytes(pkt[IP].payload))
+    first.len = len(buffer.getvalue()) + 20
+    first.flags = 0
+    del(first.chksum)
+    if return_ip:
+        header = bytes(first[IP])[:20]
+        return first[IP].__class__(header + buffer.getvalue())
+    else:
+        header = bytes(first[Ether])[:34]
+        return first[Ether].__class__(header + buffer.getvalue())
+
+
+def reassemble4_ether(listoffragments):
+    return reassemble4_core(listoffragments, False)
+
+
+def reassemble4(listoffragments):
+    return reassemble4_core(listoffragments, True)