If is_finished() cannot be converted into get_finish_string(), i.e.
is_finished() is a condition that depends on multiple output lines make sure
we call super().is_finished(line) on every line so that we catch Assertion
errors etc. that are checked for in TestCommon.is_finished().
Signed-off-by: Simon Gerber <simon.gerber@inf.ethz.ch>
self.saw_line_A = True
if line.startswith(self.get_test_B_finish_string()):
self.saw_line_B = True
- if self.saw_line_A and self.saw_line_B:
- return True
- return False
+ return (self.saw_line_A and self.saw_line_B) or \
+ super(TimerTest, self).is_finished(line)
if test_A_passed and test_B_passed:
return PassFailResult(True)
- if test_A_passed and test_B_passed:
- return PassFailResult(True)
- else:
- return PassFailResult(False)
+ return PassFailResult(test_A_passed and test_B_passed)
def setup(self, build, machine, testdir):
super(NetCommon, self).setup(build, machine, testdir)
self.testdir = testdir
- self.finished = False
self.ip = None
self.traceLogsON = False
self.traceLogs = []
if line.startswith('dump trac buffers: Start') :
self.traceLogsON = True
elif line.startswith('dump trac buffers: Stop') :
- self.finished = True
self.traceLogsON = False
elif self.traceLogsON :
self.traceLogs.append(line);
-
- def is_finished(self, line):
- return self.finished
-
+ def get_finish_string(self):
+ return 'dump trac buffers: Stop'
@tests.add_test
class NetdTraceTest(NetCommon):
return modules
- def is_finished(self, line):
- return line.startswith("AHCI testing completed.")
+ def get_finish_string(self):
+ return "AHCI testing completed."
def process_data(self, testdir, rawiter):
self.regex = re.compile(self.REGEX)
def is_finished(self, line):
if line.startswith("retype: result:"):
self._nseen += 1
- return self._nseen == self._ncores
+ return self._nseen == self._ncores or super(RetypeMultiTest, self).is_finished(line)
def process_data(self, testdir, rawiter):
# the test passed iff the last line is the finish string
return modules
- def is_finished(self, line):
- return line.startswith("SUCCESS")
+ def get_finish_string(self):
+ return "SUCCESS"
def process_data(self, testdir, rawiter):
passed = False
self.finished = True
def is_finished(self, line):
- return self.finished
+ return self.finished or super(EchoTestCommon, self).is_finished(line)
def get_ipbench_test(self):
return (IPBENCH_TEST, IPBENCH_TEST_ARGS)
modules.add_module("e1000n_irqtest", ["auto"])
return modules
- def is_finished(self, line):
- return line.startswith("TEST ")
+ def get_finish_string(self):
+ return "TEST "
def process_data(self, testdir, rawiter):
for line in rawiter:
# XXX: count number of times we have seen the finish string
if line.startswith("memtest passed successfully!"):
self._nseen += 1
- return self._nseen == self._ncores
+ return self._nseen == self._ncores or super(MemTestMulti, self).is_finished(line)
def process_data(self, testdir, rawiter):
nspawned = nseen = 0
modules.add_module("mt_waitset", ["10", "10", "10000"])
return modules
- def is_finished(self, line):
- return "Test PASSED" in line or "Test FAILED" in line
+ def get_finish_string(self):
+ # Finish line is "Test PASSED" or "Test FAILED"
+ return "Test "
def process_data(self, testdir, rawiter):
passed = False
def setup(self, build, machine, testdir):
super(SpanTest, self).setup(build, machine, testdir)
- # XXX: track number of cores booted and seen for is_finished()
- self._ncores = machine.get_ncores()
- self._nseen = 0
-
def get_modules(self, build, machine):
modules = super(SpanTest, self).get_modules(build, machine)
# span on all cores other than 0 -- matches spantest code
return modules
def is_finished(self, line):
- return re.match(MATCH, line)
+ return re.match(MATCH, line) or super(SpanTest, self).is_finished(line)
def process_data(self, testdir, rawiter):
result = False
def setup(self, build, machine, testdir):
super(SpanTestInterleaved, self).setup(build, machine, testdir)
- # XXX: track number of cores booted and seen for is_finished()
- self._ncores = machine.get_ncores()
- self._nseen = 0
-
def get_modules(self, build, machine):
modules = super(SpanTestInterleaved, self).get_modules(build, machine)
# span on all cores other than 0 -- matches spantest code
modules.add_module("tests/span-interleaved", [ machine.get_ncores() ])
return modules
- def is_finished(self, line):
- return re.search('SPAN_TEST_SUCCESS.', line)
+ def get_finish_string(self):
+ return 'SPAN_TEST_SUCCESS.'
def process_data(self, testdir, rawiter):
result = False
def setup(self, build, machine, testdir):
super(SpanTestExit, self).setup(build, machine, testdir)
- # XXX: track number of cores booted and seen for is_finished()
- self._ncores = machine.get_ncores()
- self._nseen = 0
-
def get_modules(self, build, machine):
modules = super(SpanTestExit, self).get_modules(build, machine)
# span on all cores other than 0 -- matches spantest code
def is_finished(self, line):
if re.search('SPAN_TEST_DONE.', line) :
self.is_done = True
- return re.match('kernel [0-9]*: user page fault WHILE DISABLED', line)
+ return re.match('kernel [0-9]*: user page fault WHILE DISABLED', line) or \
+ super(SpanTestExit, self).is_finished(line)
def process_data(self, testdir, rawiter):
result = True
def is_finished(self, line):
if re.match(MATCH, line):
self._nseen += 1
- return self._nseen == (NUM_SPAWNS + 1)
+ return self._nseen == (NUM_SPAWNS + 1) or super(SpawnTest, self).is_finished(line)
def process_data(self, testdir, rawiter):
nspawned = 0
return len(self.server_failures) == 0
def is_finished(self, line):
- return self.finished
+ return self.finished or super(WebCommon, self).is_finished(line)
@tests.add_test
return modules
- def is_finished(self, line):
- #m = re.search("Xeon Phi operational: xeon_phi.([0-9]).ready", line)
- m = re.search("Xeon Phi operational: xeon_phi." + str(self.nphi - 1) + ".ready", line)
- if m :
- return True
- else :
- return False
-
def get_finish_string(self):
- return "Xeon Phi operational: xeon_phi.([0-9]).ready"
+ return "Xeon Phi operational: xeon_phi." + str(self.nphi - 1) + ".ready"
def boot(self, *args):
super(XeonPhi_Boot_Test, self).boot(*args)