qemu-devel
[Top][All Lists]
Advanced

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

[Qemu-devel] [PATCH V2 2/6] runner: Tool for fuzz tests execution


From: Maria Kustova
Subject: [Qemu-devel] [PATCH V2 2/6] runner: Tool for fuzz tests execution
Date: Fri, 4 Jul 2014 15:39:29 +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.

v1 -> v2:
Added parameter for a fuzzer configuration file

Signed-off-by: Maria Kustova <address@hidden>
---
 tests/image-fuzzer/runner/runner.py | 290 ++++++++++++++++++++++++++++++++++++
 1 file changed, 290 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..bf28300
--- /dev/null
+++ b/tests/image-fuzzer/runner/runner.py
@@ -0,0 +1,290 @@
+#!/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 ConfigParser
+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, fuzz_config=None):
+        """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
+        self.fuzz_config = fuzz_config
+
+    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 the 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 an image generation call
+        # as posssible to avoid a corruption of the random sequence
+        random.seed(self.seed)
+        image_generator.create_image('test_image', self.fuzz_config)
+        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
+          --config=FILE                 take fuzzer a configuration from FILE
+          -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, fuzz_config):
+        """Setup environment for one test and execute this test"""
+        try:
+            test = TestEnv(test_id, seed, work_dir, run_log, test_bin, cleanup,
+                           log_all, fuzz_config)
+        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=',
+                                        'config=', '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
+    config_file = 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
+        elif opt == '--config':
+            config_file = 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'
+    # Read fuzzer configuration
+    if config_file is None:
+        fuzz_config = None
+    else:
+        fuzz_config = []
+        config = ConfigParser.RawConfigParser()
+        config.read(config_file)
+        for s in config.sections():
+            opt_list = config.options(s)
+            if len(opt_list) == 0:
+                fuzz_config.append([s])
+            else:
+                for o in opt_list:
+                    fuzz_config.append([s, o])
+    # 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, fuzz_config)
+        except (KeyboardInterrupt, SystemExit):
+            sys.exit(1)
+
+        if seed is not None:
+            break
-- 
1.9.3




reply via email to

[Prev in Thread] Current Thread [Next in Thread]