summaryrefslogtreecommitdiffstats
path: root/test/util.py
diff options
context:
space:
mode:
authorOle Troan <ot@cisco.com>2018-12-06 17:35:12 +0100
committerNeale Ranns <nranns@cisco.com>2018-12-10 08:01:56 +0000
commit7f99183a20d60cd5c648cc23d7a2f30a594a215b (patch)
treebb3934aab296ce5fa531bfc5ab47921a113ffd26 /test/util.py
parent521a8d7df423a0b5aaf259d49ca9230705bc25ee (diff)
Test framework: StringIO fixes for Python3
Add 2/3 support to binarytomac and mactobinary and move to vpp_mac.py Change-Id: I3dc7e4a24486aee22140c781aae7e44e58935877 Signed-off-by: Ole Troan <ot@cisco.com>
Diffstat (limited to 'test/util.py')
-rw-r--r--test/util.py36
1 files changed, 29 insertions, 7 deletions
diff --git a/test/util.py b/test/util.py
index 1ab5c1f2350..a3ec6e3326b 100644
--- a/test/util.py
+++ b/test/util.py
@@ -4,7 +4,6 @@ import socket
import sys
import os.path
from abc import abstractmethod, ABCMeta
-from cStringIO import StringIO
from scapy.utils6 import in6_mactoifaceid
from scapy.layers.l2 import Ether
@@ -14,11 +13,13 @@ from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
IPv6ExtHdrHopByHop
from scapy.utils import hexdump
from socket import AF_INET6
+from io import BytesIO
+from vpp_mac import mactobinary
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)
@@ -58,11 +59,6 @@ def ip4n_range(ip4n, 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
@@ -441,3 +437,29 @@ def fragment_rfc8200(packet, identification, fragsize, _logger=None):
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)