[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-devel] [PATCH 2/5] runner: Tool for fuzz tests execution
From: |
Maria Kustova |
Subject: |
[Qemu-devel] [PATCH 2/5] runner: Tool for fuzz tests execution |
Date: |
Mon, 30 Jun 2014 15:48:37 +0400 |
The purpose of the test runner is to prepare test environment (e.g. create a
work directory, a test image, etc), execute the program under test with
parameters, indicate a test failure if the program was killed during test
execution and collect core dumps, logs and other test artifacts.
The test runner doesn't depend on image format or a program will be tested, so
it can be used with any external image generator and program under test.
Signed-off-by: Maria Kustova <address@hidden>
---
tests/image-fuzzer/runner/runner.py | 270 ++++++++++++++++++++++++++++++++++++
1 file changed, 270 insertions(+)
create mode 100755 tests/image-fuzzer/runner/runner.py
diff --git a/tests/image-fuzzer/runner/runner.py
b/tests/image-fuzzer/runner/runner.py
new file mode 100755
index 0000000..21de78e
--- /dev/null
+++ b/tests/image-fuzzer/runner/runner.py
@@ -0,0 +1,270 @@
+#!/usr/bin/env python
+
+# Tool for running fuzz tests
+#
+# Copyright (C) 2014 Maria Kustova <address@hidden>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+
+import sys, os, signal
+from time import time
+import subprocess
+import random
+from itertools import count
+from shutil import rmtree
+import getopt
+import resource
+resource.setrlimit(resource.RLIMIT_CORE, (-1, -1))
+
+
+def multilog(msg, *output):
+ """ Write an object to all of specified file descriptors
+ """
+
+ for fd in output:
+ fd.write(msg)
+ fd.flush()
+
+
+def str_signal(sig):
+ """ Convert a numeric value of a system signal to the string one
+ defined by the current operational system
+ """
+
+ for k, v in signal.__dict__.items():
+ if v == sig:
+ return k
+
+
+class TestException(Exception):
+ """Exception for errors risen by TestEnv objects"""
+ pass
+
+
+class TestEnv(object):
+ """ Trivial test object
+
+ The class sets up test environment, generates a test image and executes
+ application under tests with specified arguments and a test image provided.
+ All logs are collected.
+ Summary log will contain short descriptions and statuses of tests in
+ a run.
+ Test log will include application (e.g. 'qemu-img') logs besides info sent
+ to the summary log.
+ """
+
+ def __init__(self, test_id, seed, work_dir, run_log, exec_bin=None,
+ cleanup=True, log_all=False):
+ """Set test environment in a specified work directory.
+
+ Path to qemu_img will be retrieved from 'QEMU_IMG' environment
+ variable, if a test binary is not specified.
+ """
+
+ if seed is not None:
+ self.seed = seed
+ else:
+ self.seed = hash(time())
+
+ self.init_path = os.getcwd()
+ self.work_dir = work_dir
+ self.current_dir = os.path.join(work_dir, 'test-' + test_id)
+ if exec_bin is not None:
+ self.exec_bin = exec_bin.strip().split(' ')
+ else:
+ self.exec_bin = \
+ os.environ.get('QEMU_IMG', 'qemu-img').strip().split(' ')
+
+ try:
+ os.makedirs(self.current_dir)
+ except OSError:
+ e = sys.exc_info()[1]
+ print >>sys.stderr, \
+ "Error: The working directory '%s' cannot be used. Reason: %s"\
+ % (self.work_dir, e[1])
+ raise TestException
+ self.log = open(os.path.join(self.current_dir, "test.log"), "w")
+ self.parent_log = open(run_log, "a")
+ self.result = False
+ self.cleanup = cleanup
+ self.log_all = log_all
+
+ def _test_app(self, q_args):
+ """ Start application under test with specified arguments and return
+ an exit code or a kill signal depending on result of an execution.
+ """
+ devnull = open('/dev/null', 'r+')
+ return subprocess.call(self.exec_bin + q_args,
+ stdin=devnull, stdout=self.log, stderr=self.log)
+
+ def execute(self, q_args):
+ """ Execute a test.
+
+ The method creates a test image, runs test app and analyzes its exit
+ status. If the application was killed by a signal, the test is marked
+ as failed.
+ """
+ os.chdir(self.current_dir)
+ # Seed initialization should be as close to image generation call
+ # as posssible to avoid a corruption of random sequence
+ random.seed(self.seed)
+ image_generator.create_image('test_image')
+ test_summary = "Seed: %s\nCommand: %s\nTest directory: %s\n" \
+ % (self.seed, " ".join(q_args), self.current_dir)
+ try:
+ retcode = self._test_app(q_args)
+ except OSError:
+ e = sys.exc_info()[1]
+ multilog(test_summary + "Error: Start of '%s' failed. " \
+ "Reason: %s\n\n" % (os.path.basename(self.exec_bin[0]),
+ e[1]),
+ sys.stderr, self.log, self.parent_log)
+ raise TestException
+
+ if retcode < 0:
+ multilog(test_summary + "FAIL: Test terminated by signal %s\n\n"
+ % str_signal(-retcode), sys.stderr, self.log,
+ self.parent_log)
+ elif self.log_all:
+ multilog(test_summary + "PASS: Application exited with the code" +
+ " '%d'\n\n" % retcode, sys.stdout, self.log,
+ self.parent_log)
+ self.result = True
+ else:
+ self.result = True
+
+ def finish(self):
+ """ Restore environment after a test execution. Remove folders of
+ passed tests
+ """
+ self.log.close()
+ self.parent_log.close()
+ os.chdir(self.init_path)
+ if self.result and self.cleanup:
+ rmtree(self.current_dir)
+
+if __name__ == '__main__':
+
+ def usage():
+ print """
+ Usage: runner.py [OPTION...] TEST_DIR IMG_GENERATOR
+
+ Set up test environment in TEST_DIR and run a test in it. A module for
+ test image generation should be specified via IMG_GENERATOR. Use
+ TEST_IMG alias to mark the position in the command where a test image
+ name should be placed.
+ Example:
+ python runner.py -b ./qemu-img -c 'info TEST_IMG' /tmp/test ../qcow2
+
+ Optional arguments:
+ -h, --help display this help and exit
+ -b, --binary=PATH path to the application under test,
+ by default "qemu-img" in PATH or
+ QEMU_IMG environment variables
+ -c, --command=STRING execute the tested application
+ with arguments specified,
+ by default STRING="check"
+ -s, --seed=STRING seed for a test image generation,
+ by default will be generated randomly
+ -k, --keep_passed don't remove folders of passed tests
+ -v, --verbose log information about passed tests
+ """
+
+ def run_test(test_id, seed, work_dir, run_log, test_bin, cleanup, log_all,
+ command):
+ """Setup environment for one test and execute this test"""
+ try:
+ test = TestEnv(test_id, seed, work_dir, run_log, test_bin, cleanup,
+ log_all)
+ except TestException:
+ sys.exit(1)
+
+ # Python 2.4 doesn't support 'finally' and 'except' in the same 'try'
+ # block
+ try:
+ try:
+ test.execute(command)
+ # Silent exit on user break
+ except TestException:
+ sys.exit(1)
+ finally:
+ test.finish()
+
+ try:
+ opts, args = getopt.gnu_getopt(sys.argv[1:], 'c:hb:s:kv',
+ ['command=', 'help', 'binary=', 'seed=',
+ 'keep_passed', 'verbose'])
+ except getopt.error:
+ e = sys.exc_info()[1]
+ print "Error: %s\n\nTry 'runner.py --help' for more information" % e
+ sys.exit(1)
+
+ command = ['check']
+ cleanup = True
+ log_all = False
+ test_bin = None
+ seed = None
+ for opt, arg in opts:
+ if opt in ('-h', '--help'):
+ usage()
+ sys.exit()
+ elif opt in ('-c', '--command'):
+ command = arg.split(" ")
+ elif opt in ('-k', '--keep_passed'):
+ cleanup = False
+ elif opt in ('-v', '--verbose'):
+ log_all = True
+ elif opt in ('-b', '--binary'):
+ test_bin = os.path.realpath(arg)
+ elif opt in ('-s', '--seed'):
+ seed = arg
+
+ if not len(args) == 2:
+ print "Missed parameter\nTry 'runner.py --help' " \
+ "for more information"
+ sys.exit(1)
+
+ work_dir = os.path.realpath(args[0])
+ # run_log is created in 'main', because multiple tests are expected to
+ # log in it
+ run_log = os.path.join(work_dir, 'run.log')
+
+ # Add the path to the image generator module to sys.path
+ sys.path.append(os.path.dirname(os.path.realpath(args[1])))
+ # Remove a script extension from image generator module if any
+ generator_name = os.path.splitext(os.path.basename(args[1]))[0]
+ try:
+ image_generator = __import__(generator_name)
+ except ImportError:
+ e = sys.exc_info()[1]
+ print "Error: The image generator '%s' cannot be imported.\n" \
+ "Reason: %s" % (generator_name, e)
+ sys.exit(1)
+
+ # Replace test image alias with its real name
+ for idx, item in enumerate(command):
+ if item == 'TEST_IMG':
+ command[idx] = 'test_image'
+ # If a seed is specified, only one test will be executed.
+ # Otherwise runner will terminate after a keyboard interruption
+ for test_id in count(1):
+ try:
+ run_test(str(test_id), seed, work_dir, run_log, test_bin, cleanup,
+ log_all, command)
+ except (KeyboardInterrupt, SystemExit):
+ sys.exit(1)
+
+ if seed is not None:
+ break
--
1.9.3