from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
getLogger, colorize
from vpp_object import VppObjectRegistry
+from util import ppp
+from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
+from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
+from scapy.layers.inet6 import ICMPv6EchoReply
if os.name == 'posix' and sys.version_info[0] < 3:
# using subprocess32 is recommended by python official documentation
# @ https://docs.python.org/2/library/subprocess.html
else:
import subprocess
+
debug_framework = False
if os.getenv('TEST_DEBUG', "0") == "1":
debug_framework = True
print("Now is the time to attach a gdb by running the above "
"command and set up breakpoints etc.")
print(single_line_delim)
- sys.stdin.readline("Press ENTER to continue running the testcase...")
+ raw_input("Press ENTER to continue running the testcase...")
@classmethod
def run_vpp(cls):
print(double_line_delim)
print("VPP or GDB server is still running")
print(single_line_delim)
- sys.stdin.readline(
- "When done debugging, press ENTER to kill the process and "
- "finish running the testcase...")
+ raw_input("When done debugging, press ENTER to kill the "
+ "process and finish running the testcase...")
os.write(cls.pump_thread_wakeup_pipe[1], 'ding dong wake up')
cls.pump_thread_stop_flag.set()
self.logger.info(self.vapi.ppcli("show hardware"))
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show run"))
+ self.logger.info(self.vapi.ppcli("show log"))
self.registry.remove_vpp_config(self.logger)
# Save/Dump VPP api trace log
api_trace = "vpp_api_trace.%s.log" % self._testMethodName
return result
@classmethod
- def create_loopback_interfaces(cls, interfaces):
+ def create_loopback_interfaces(cls, count):
"""
Create loopback interfaces.
- :param interfaces: iterable indexes of the interfaces.
+ :param count: number of interfaces created.
:returns: List of created interfaces.
"""
- result = []
- for i in interfaces:
- intf = VppLoInterface(cls, i)
+ result = [VppLoInterface(cls) for i in range(count)]
+ for intf in result:
setattr(cls, intf.name, intf)
- result.append(intf)
cls.lo_interfaces = result
return result
name, real_value, expected_min, expected_max)
self.assertTrue(expected_min <= real_value <= expected_max, msg)
+ def assert_packet_checksums_valid(self, packet,
+ ignore_zero_udp_checksums=True):
+ received = packet.__class__(str(packet))
+ self.logger.debug(
+ ppp("Verifying packet checksums for packet:", received))
+ udp_layers = ['UDP', 'UDPerror']
+ checksum_fields = ['cksum', 'chksum']
+ checksums = []
+ counter = 0
+ temp = received.__class__(str(received))
+ while True:
+ layer = temp.getlayer(counter)
+ if layer:
+ for cf in checksum_fields:
+ if hasattr(layer, cf):
+ if ignore_zero_udp_checksums and \
+ 0 == getattr(layer, cf) and \
+ layer.name in udp_layers:
+ continue
+ delattr(layer, cf)
+ checksums.append((counter, cf))
+ else:
+ break
+ counter = counter + 1
+ if 0 == len(checksums):
+ return
+ temp = temp.__class__(str(temp))
+ for layer, cf in checksums:
+ calc_sum = getattr(temp[layer], cf)
+ self.assert_equal(
+ getattr(received[layer], cf), calc_sum,
+ "packet checksum on layer #%d: %s" % (layer, temp[layer].name))
+ self.logger.debug(
+ "Checksum field `%s` on `%s` layer has correct value `%s`" %
+ (cf, temp[layer].name, calc_sum))
+
+ def assert_checksum_valid(self, received_packet, layer,
+ field_name='chksum',
+ ignore_zero_checksum=False):
+ """ Check checksum of received packet on given layer """
+ received_packet_checksum = getattr(received_packet[layer], field_name)
+ if ignore_zero_checksum and 0 == received_packet_checksum:
+ return
+ recalculated = received_packet.__class__(str(received_packet))
+ delattr(recalculated[layer], field_name)
+ recalculated = recalculated.__class__(str(recalculated))
+ self.assert_equal(received_packet_checksum,
+ getattr(recalculated[layer], field_name),
+ "packet checksum on layer: %s" % layer)
+
+ def assert_ip_checksum_valid(self, received_packet,
+ ignore_zero_checksum=False):
+ self.assert_checksum_valid(received_packet, 'IP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_tcp_checksum_valid(self, received_packet,
+ ignore_zero_checksum=False):
+ self.assert_checksum_valid(received_packet, 'TCP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_udp_checksum_valid(self, received_packet,
+ ignore_zero_checksum=True):
+ self.assert_checksum_valid(received_packet, 'UDP',
+ ignore_zero_checksum=ignore_zero_checksum)
+
+ def assert_embedded_icmp_checksum_valid(self, received_packet):
+ if received_packet.haslayer(IPerror):
+ self.assert_checksum_valid(received_packet, 'IPerror')
+ if received_packet.haslayer(TCPerror):
+ self.assert_checksum_valid(received_packet, 'TCPerror')
+ if received_packet.haslayer(UDPerror):
+ self.assert_checksum_valid(received_packet, 'UDPerror',
+ ignore_zero_checksum=True)
+ if received_packet.haslayer(ICMPerror):
+ self.assert_checksum_valid(received_packet, 'ICMPerror')
+
+ def assert_icmp_checksum_valid(self, received_packet):
+ self.assert_checksum_valid(received_packet, 'ICMP')
+ self.assert_embedded_icmp_checksum_valid(received_packet)
+
+ def assert_icmpv6_checksum_valid(self, pkt):
+ if pkt.haslayer(ICMPv6DestUnreach):
+ self.assert_checksum_valid(pkt, 'ICMPv6DestUnreach', 'cksum')
+ self.assert_embedded_icmp_checksum_valid(pkt)
+ if pkt.haslayer(ICMPv6EchoRequest):
+ self.assert_checksum_valid(pkt, 'ICMPv6EchoRequest', 'cksum')
+ if pkt.haslayer(ICMPv6EchoReply):
+ self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
+
@classmethod
def sleep(cls, timeout, remark=None):
if hasattr(cls, 'logger'):
"Finished sleep (%s) - slept %ss (wanted %ss)" % (
remark, after - before, timeout))
- def send_and_assert_no_replies(self, intf, pkts, remark=""):
+ def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None):
self.vapi.cli("clear trace")
intf.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- timeout = 1
+ if not timeout:
+ timeout = 1
for i in self.pg_interfaces:
i.get_capture(0, timeout=timeout)
i.assert_nothing_captured(remark=remark)
if hasattr(self, 'test_framework_failed_pipe'):
pipe = self.test_framework_failed_pipe
if pipe:
- pipe.send(test.__class__)
+ if test.__class__.__name__ == "_ErrorHolder":
+ x = str(test)
+ if x.startswith("setUpClass"):
+ # x looks like setUpClass (test_function.test_class)
+ cls = x.split(".")[1].split(")")[0]
+ for t in self.test_suite:
+ if t.__class__.__name__ == cls:
+ pipe.send(t.__class__)
+ break
+ else:
+ raise Exception("Can't find class name `%s' "
+ "(from ErrorHolder) in test suite "
+ "`%s'" % (cls, self.test_suite))
+ else:
+ raise Exception("FIXME: unexpected special case - "
+ "ErrorHolder description is `%s'" %
+ str(test))
+ else:
+ pipe.send(test.__class__)
def addFailure(self, test, err):
"""
filtered.countTestCases(), test.countTestCases()))
if not running_extended_tests():
print("Not running extended tests (some tests will be skipped)")
+ # super-ugly hack #2
+ VppTestResult.test_suite = filtered
return super(VppTestRunner, self).run(filtered)
Code Review / vpp.git / blobdiff