ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Test framework: StringIO fixes for Python3

Browse Source 
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>
This commit is contained in:
Ole Troan
2018-12-06 17:35:12 +01:00
committed by Neale Ranns
parent 521a8d7df4
commit 7f99183a20
23 changed files with 124 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
test/util.py
View File

@ -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)