""" pump output from vpp stdout/stderr to proper queues """
stdout_fragment = ""
stderr_fragment = ""
- while not testclass.pump_thread_stop_flag.wait(0):
+ while not testclass.pump_thread_stop_flag.is_set():
readable = select.select([testclass.vpp.stdout.fileno(),
testclass.vpp.stderr.fileno(),
testclass.pump_thread_wakeup_pipe[0]],
coredump_size, "}", "api-trace", "{", "on", "}",
"api-segment", "{", "prefix", cls.shm_prefix, "}",
"plugins", "{", "plugin", "dpdk_plugin.so", "{",
- "disable", "}", "}", ]
+ "disable", "}", "plugin", "unittest_plugin.so",
+ "{", "enable", "}", "}", ]
if plugin_path is not None:
cls.vpp_cmdline.extend(["plugin_path", plugin_path])
cls.logger.info("vpp_cmdline: %s" % cls.vpp_cmdline)
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()
+ os.write(cls.pump_thread_wakeup_pipe[1], 'ding dong wake up')
if hasattr(cls, 'pump_thread'):
cls.logger.debug("Waiting for pump thread to stop")
cls.pump_thread.join()
"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)