#!/usr/bin/env python
###
### Copyright (C) 2011 Texas Instruments
###
### Licensed under the Apache License, Version 2.0 (the "License");
### you may not use this file except in compliance with the License.
### You may obtain a copy of the License at
###
### http://www.apache.org/licenses/LICENSE-2.0
###
### Unless required by applicable law or agreed to in writing, software
### distributed under the License is distributed on an "AS IS" BASIS,
### WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
### See the License for the specific language governing permissions and
### limitations under the License.
###
"""TestFlinger meta-test execution framework
When writing a master test script that runs several scripts, this module
can be used to execute those tests in a detached process (sandbox).
Thus, if the test case fails by a segfault or timeout, this can be
detected and the upper-level script simply moves on to the next script.
"""
import os
import time
import subprocess
import sys
import time
g_default_timeout = 300
class TestCase:
"""Test running wrapper object."""
def __init__(self, TestDict = {}, Logfile = None):
"""Set up the test runner object.
TestDict: dictionary with the test properties. (string: value). The
recognized properties are:
filename - name of executable test file
Type: string
Required: yes
args - command line arguments for test
Type: list of strings, or None
Required: no
Default: None
timeout - upper limit on execution time (secs). If test takes
this long to run, then it is deemed a failure
Type: integer
Required: no
Default: TestFlinger.g_default_timeout (typ. 300 sec)
expect-fail - If the test is expected to fail (return non-zero)
in order to pass, set this to True
Type: bool
Required: no
Default: False
expect-signal If the test is expected to fail because of
a signal (e.g. SIGTERM, SIGSEGV) then this is considered success
Type: bool
Required: no
Default: False
Logfile: a file object where stdout/stderr for the tests should be dumped.
If null, then no logging will be done. (See also TestFlinger.setup_logfile()
and TestFlinger.close_logfile().
"""
global g_default_timeout
self._program = None
self._args = None
self._timeout = g_default_timeout # Default timeout
self._verdict = None
self._expect_fail = False
self._expect_signal = False
self._logfile = Logfile
self._proc = None
self._time_expire = None
self._program = TestDict['filename']
if 'args' in TestDict:
self._args = TestDict['args']
if 'timeout' in TestDict and TestDict['timeout'] is not None:
self._timeout = TestDict['timeout']
if 'expect-fail' in TestDict and TestDict['expect-fail'] is not None:
self._expect_fail = TestDict['expect-fail']
if 'expect-signal' in TestDict and TestDict['expect-signal'] is not None:
self._expect_signal = TestDict['expect-signal']
def __del__(self):
pass
def start(self):
"""Starts the test in another process. Returns True if the
test was successfully spawned. False if there was an error.
"""
command = os.path.abspath(self._program)
if not os.path.exists(command):
print "ERROR: The program to execute does not exist (%s)" % (command,)
return False
timestamp = time.strftime("%Y.%m.%d %H:%M:%S")
now = time.time()
self._time_expire = self._timeout + now
self._kill_timeout = False
self._log_write("====================================================================\n")
self._log_write("BEGINNG TEST '%s' at %s\n" % (self._program, timestamp))
self._log_write("--------------------------------------------------------------------\n")
self._log_flush()
self._proc = subprocess.Popen(args=command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
return (self._proc is not None)
def wait(self):
"""Blocks until the test completes or times out, whichever
comes first. If test fails, returns False. Otherwise returns
true.
"""
if self._proc is None:
print "ERROR: Test was never started"
return False
self._proc.poll()
while (time.time() < self._time_expire) and (self._proc.poll() is None):
self._process_logs()
time.sleep(.5)
if self._proc.returncode is None:
self.kill()
return False
self._process_logs()
self._finalize_log()
return True
def kill(self):
"""Kill the currently running test (if there is one).
"""
if self._proc is None:
print "WARNING: killing a test was never started"
return False
self._kill_timeout = True
self._proc.terminate()
time.sleep(2)
self._proc.kill()
self._log_write("\nKilling process by request...\n")
self._log_flush()
self._finalize_log()
return True
def verdict(self):
"""Returns a string, either 'PASS', 'FAIL', 'FAIL/TIMEOUT', or 'FAIL/SIGNAL(n)
'"""
self._proc.poll()
rc = self._proc.returncode
if rc is None:
print "ERROR: test is still running"
if self._kill_timeout:
return "FAIL/TIMOUT"
if rc < 0 and self._expect_signal:
return "PASS"
elif rc < 0:
return "FAIL/SIGNAL(%d)" % (-rc,)
if self._expect_fail:
if rc != 0:
return "PASS"
else:
return "FAIL"
else:
if rc == 0:
return "PASS"
else:
return "FAIL"
def _process_logs(self):
if self._logfile is not None:
data = self._proc.stdout.read()
self._logfile.write(data)
self._logfile.flush()
def _finalize_log(self):
timestamp = time.strftime("%Y.%m.%d %H:%M:%S")
self._log_write("--------------------------------------------------------------------\n")
self._log_write("ENDING TEST '%s' at %s\n" % (self._program, timestamp))
self._log_write("====================================================================\n")
self._log_flush()
def _log_write(self, data):
if self._logfile is not None:
self._logfile.write(data)
def _log_flush(self):
if self._logfile is not None:
self._logfile.flush()
def setup_logfile(override_logfile_name = None):
"""Open a logfile and prepare it for use with TestFlinger logging.
The filename will be generated based on the current date/time.
If override_logfile_name is not None, then that filename will be
used instead.
See also: close_logfile()
"""
tmpfile = None
if override_logfile_name is not None:
tmpfile = override_logfile_name
if os.path.exists(tmpfile):
os.unlink(tmpfile)
else:
tmpfile = time.strftime("test-log-%Y.%m.%d.%H%M%S.txt")
while os.path.exists(tmpfile):
tmpfile = time.strftime("test-log-%Y.%m.%d.%H%M%S.txt")
fobj = open(tmpfile, 'wt')
print "Logging to", tmpfile
timestamp = time.strftime("%Y.%m.%d %H:%M:%S")
fobj.write("BEGINNING TEST SET %s\n" % (timestamp,))
fobj.write("====================================================================\n")
return fobj
def close_logfile(fobj):
"""Convenience function for closing a TestFlinger log file.
fobj: an open and writeable file object
See also : setup_logfile()
"""
timestamp = time.strftime("%Y.%m.%d %H:%M:%S")
fobj.write("====================================================================\n")
fobj.write("CLOSING TEST SET %s\n" % (timestamp,))
if __name__ == "__main__":
pass