summary |
shortlog |
log |
commit | commitdiff |
review |
tree
raw |
patch |
inline | side by side (from parent 1:
40cfc15)
Raise a new UnexpectedPacketErrror, when a packet is captured
unexpectedly. This pretty-prints a terse description of said packet.
Type: improvement
Signed-off-by: Klement Sekera <klement.sekera@gmail.com>
Change-Id: Ibac19fc5bbd82a150fec3c90940a37af6344fd4f
if not timeout:
timeout = 1
for i in self.pg_interfaces:
if not timeout:
timeout = 1
for i in self.pg_interfaces:
- i.get_capture(0, timeout=timeout)
- i.assert_nothing_captured(remark=remark)
+ i.assert_nothing_captured(timeout=timeout, remark=remark)
timeout = 0.1
finally:
if trace:
timeout = 0.1
finally:
if trace:
timeout = 1
for i in self.pg_interfaces:
if i not in outputs:
timeout = 1
for i in self.pg_interfaces:
if i not in outputs:
- i.get_capture(0, timeout=timeout)
- i.assert_nothing_captured()
+ i.assert_nothing_captured(timeout=timeout)
timeout = 0.1
if stats_diff:
timeout = 0.1
if stats_diff:
from socket import AF_INET6
import os.path
from copy import deepcopy
from socket import AF_INET6
import os.path
from copy import deepcopy
+from collections import UserDict
import scapy.compat
from scapy.layers.l2 import Ether
import scapy.compat
from scapy.layers.l2 import Ether
null_logger.addHandler(logging.NullHandler())
null_logger.addHandler(logging.NullHandler())
+def pr(packet):
+ return packet.__repr__()
+
+
def ppp(headline, packet):
def ppp(headline, packet):
- """ Return string containing the output of scapy packet.show() call. """
+ """ Return string containing headline and output of scapy packet.show() """
return '%s\n%s\n\n%s\n' % (headline,
hexdump(packet, dump=True),
packet.show(dump=True))
return '%s\n%s\n\n%s\n' % (headline,
hexdump(packet, dump=True),
packet.show(dump=True))
return reassemble4_core(listoffragments, True)
return reassemble4_core(listoffragments, True)
+class UnexpectedPacketError(Exception):
+ def __init__(self, packet, msg=""):
+ self.packet = packet
+ self.msg = msg
+
+ def __str__(self):
+ return f"\nUnexpected packet:\n{pr(self.packet)}{self.msg}"
+
+
def recursive_dict_merge(dict_base, dict_update):
"""Recursively merge base dict with update dict, return merged dict"""
for key in dict_update:
def recursive_dict_merge(dict_base, dict_update):
"""Recursively merge base dict with update dict, return merged dict"""
for key in dict_update:
+class StatsDiff(UserDict):
"""
Diff dictionary is a dictionary of dictionaries of interesting stats:
"""
Diff dictionary is a dictionary of dictionaries of interesting stats:
- def __init__(self, stats_diff={}):
- self.stats_diff = stats_diff
+ __slots__ = () # prevent setting properties to act like a dictionary
- def update(self, sw_if_index, key, value):
- if sw_if_index in self.stats_diff:
- self.stats_diff[sw_if_index][key] = value
- else:
- self.stats_diff[sw_if_index] = {key: value}
+ def __init__(self, data):
+ super().__init__(data)
- return recursive_dict_merge(deepcopy(self.stats_diff), other)
+ return recursive_dict_merge(deepcopy(self.data), other)
from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA,\
ICMPv6NDOptSrcLLAddr, ICMPv6NDOptDstLLAddr, ICMPv6ND_RA, RouterAlert, \
IPv6ExtHdrHopByHop
from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA,\
ICMPv6NDOptSrcLLAddr, ICMPv6NDOptDstLLAddr, ICMPv6ND_RA, RouterAlert, \
IPv6ExtHdrHopByHop
-from util import ppp, ppc
+from util import ppp, ppc, UnexpectedPacketError
from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ismaddr
from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ismaddr
remaining_time -= elapsed_time
if capture:
self.generate_debug_aid("count-mismatch")
remaining_time -= elapsed_time
if capture:
self.generate_debug_aid("count-mismatch")
+ if len(capture) > 0 and 0 == expected_count:
+ rem = f"\n{remark}" if remark else ""
+ raise UnexpectedPacketError(
+ capture[0],
+ f"\n({len(capture)} packets captured in total){rem}")
raise Exception("Captured packets mismatch, captured %s packets, "
"expected %s packets on %s" %
(len(capture.res), expected_count, name))
else:
raise Exception("No packets captured on %s" % name)
raise Exception("Captured packets mismatch, captured %s packets, "
"expected %s packets on %s" %
(len(capture.res), expected_count, name))
else:
raise Exception("No packets captured on %s" % name)
- def assert_nothing_captured(self, remark=None, filter_out_fn=is_ipv6_misc):
+ def assert_nothing_captured(self, timeout=1, remark=None,
+ filter_out_fn=is_ipv6_misc):
""" Assert that nothing unfiltered was captured on interface
:param remark: remark printed into debug logs
""" Assert that nothing unfiltered was captured on interface
:param remark: remark printed into debug logs
if os.path.isfile(self.out_path):
try:
capture = self.get_capture(
if os.path.isfile(self.out_path):
try:
capture = self.get_capture(
- 0, remark=remark, filter_out_fn=filter_out_fn)
+ 0, timeout=timeout, remark=remark,
+ filter_out_fn=filter_out_fn)
if not capture or len(capture.res) == 0:
# junk filtered out, we're good
return
if not capture or len(capture.res) == 0:
# junk filtered out, we're good
return
pass
self.generate_debug_aid("empty-assert")
if remark:
pass
self.generate_debug_aid("empty-assert")
if remark:
- raise AssertionError(
- "Non-empty capture file present for interface %s (%s)" %
- (self.name, remark))
+ raise UnexpectedPacketError(
+ capture[0],
+ f" ({len(capture)} packets captured in total) ({remark})")
- raise AssertionError("Capture file present for interface %s" %
- self.name)
+ raise UnexpectedPacketError(
+ capture[0], f" ({len(capture)} packets captured in total)")
def wait_for_pg_stop(self):
# wait till packet-generator is stopped
def wait_for_pg_stop(self):
# wait till packet-generator is stopped