Re: [Qemu-devel] [RFC] image-fuzzer: Trivial test runner

[Stefan Hajnoczi]

On Sat, May 31, 2014 at 01:56:46PM +0400, Maria Kustova wrote:

Please add a --format qcow2 (or maybe --image-generator) option using
__import__() to load the image generator module.  That way people can
drop in new image generator modules in the future and we don't hard-code
their names into the runner.

> This version of test runner executes only one test. In future it will be
> extended to execute multiple tests in a run.
> 
> Signed-off-by: Maria Kustova <address@hidden>
> ---
>  tests/image-fuzzer/runner.py | 225 
> +++++++++++++++++++++++++++++++++++++++++++
>  1 file changed, 225 insertions(+)
>  create mode 100644 tests/image-fuzzer/runner.py
> 
> diff --git a/tests/image-fuzzer/runner.py b/tests/image-fuzzer/runner.py
> new file mode 100644
> index 0000000..1dea8ef
> --- /dev/null
> +++ b/tests/image-fuzzer/runner.py
> @@ -0,0 +1,225 @@
> +# 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 3 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
> +import qcow2
> +from time import gmtime, strftime
> +import subprocess
> +from shutil import rmtree
> +import getopt
> +# -----For local test environment only
> +import resource
> +resource.setrlimit(resource.RLIMIT_CORE, (-1, -1))
> +# -----

Enabling core dumps is important.  I'm not sure why it says "For local
test environment only".

This assumes that core dumps are written to the current working
directory.  On Linux this is configurable and some distros default to a
fancier setup where core dumps are not written to the current working
directory.  For now, please add a note to the documentation explaining
that core dumps should be configured to use the current working
directory.

> +
> +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 TestEnv(object):
> +    """ Trivial test object
> +
> +    The class sets up test environment, generates a test image and executes
> +    qemu_img with specified arguments and a test image provided. All logs
> +    are collected.
> +    Summary log will contain short descriptions and statuses of all tests in
> +    a run.
> +    Test log will include application ('qemu-img') logs besides info sent
> +    to the summary log.
> +    """
> +
> +    def __init__(self, work_dir, run_log, exec_bin=None, cleanup=True):
> +        """Set test environment in a specified work directory.
> +
> +        Path to qemu_img will be retrieved from 'QEMU_IMG' environment
> +        variable, if not specified.
> +        """
> +
> +        self.init_path = os.getcwd()
> +        self.work_dir = work_dir
> +        self.current_dir = os.path.join(work_dir, 
> strftime("%Y_%m_%d_%H-%M-%S",
> +                                                           gmtime()))
> +        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 cannot be 
> used.'\
> +                ' Reason: %s' %e[1]
> +            raise Exception('Internal error')

I guess this exception is really sys.exit(1).

> +
> +        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
> +
> +    def _qemu_img(self, q_args):
> +        """ Start qemu_img 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 +
> +                               ['test_image.qcow2'], stdin=devnull,
> +                               stdout=self.log, stderr=self.log)
> +
> +
> +    def execute(self, q_args, seed, size=8*512):
> +        """ Execute a test.
> +
> +        The method creates a test image, runs 'qemu_img' 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 = qcow2.create_image('test_image.qcow2', seed, size)

The qcow2 module is missing from this patch series.

> +        multilog("Seed: %s\nCommand: %s\nTest directory: %s\n"\
> +                 %(seed, " ".join(q_args), self.current_dir),\
> +                 sys.stdout, self.log, self.parent_log)

It will probably be useful to dial back the logging for test cases that
pass.

> +        try:
> +            retcode = self._qemu_img(q_args)
> +        except OSError:
> +            e = sys.exc_info()[1]
> +            multilog("Error: Start of 'qemu_img' failed. Reason: %s\n"\
> +                     %e[1], sys.stderr, self.log, self.parent_log)
> +            raise Exception('Internal error')
> +
> +        if retcode < 0:
> +            multilog('FAIL: Test terminated by signal %s\n'
> +                     %str_signal(-retcode), sys.stderr, self.log, \
> +                     self.parent_log)
> +        else:
> +            multilog("PASS: Application exited with the code '%d'\n"
> +                     %retcode, sys.stdout, self.log, self.parent_log)
> +            self.result = True
> +
> +    def finish(self):
> +        """ Restore environment after a test execution. Remove folders of
> +        passed tests
> +        """
> +        self.log.close()
> +        # Delimiter between tests
> +        self.parent_log.write("\n")
> +        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...] DIRECTORY
> +
> +        Set up test environment in DIRECTORY and run a test in it.
> +
> +        Optional arguments:
> +          -h, --help           display this help and exit
> +          -c, --command=STRING execute qemu-img with arguments specified,
> +                               by default STRING="check"
> +          -b, --binary=PATH    path to the application under test, by default
> +                               "qemu-img" in PATH or QEMU_IMG environment
> +                               variables
> +          -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
> +        """)
> +
> +    try:
> +        opts, args = getopt.getopt(sys.argv[1:], 'c:hb:s:k',
> +                                   ['command=', 'help', 'binary=', 'seed=',
> +                                    'keep_passed'])
> +    except getopt.error:
> +        e = sys.exc_info()[1]
> +        print('Error: %s\n\nTry runner.py --help.' %e)
> +        sys.exit(1)
> +
> +    if len(sys.argv) == 1:
> +        usage()
> +        sys.exit(1)

The "if not len(args) == 1" check further down does the same thing.  It
can be confusing to mix sys.argv with getopt so I suggest dropping this
one.

> +    command = ['check']
> +    cleanup = True
> +    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 ('-b', '--binary'):
> +            test_bin = arg
> +        elif opt in ('-s', '--seed'):
> +            seed = arg
> +
> +    if not len(args) == 1:
> +        print 'Error: required parameter "DIRECTORY" missed'
> +        usage()
> +        sys.exit(1)
> +
> +    work_dir = args[0]
> +    # run_log created in 'main', because multiple tests are expected to \
> +    # log in it
> +    # TODO: Make unique run_log names on every run (for one test per run
> +    # this functionality is omitted in favor of usability)

I don't understand this TODO comment.  Here's roughly what I expected
when I read the --help output:

  work_dir/
      run.log
      failure-01/
          core
          input.img
          cmdline
          seed

> +    run_log = os.path.join(work_dir, 'run.log')
> +
> +    try:
> +        test = TestEnv(work_dir, run_log, test_bin, cleanup)
> +    except:
> +        e = sys.exc_info()[1]
> +        print("FAIL: %s"  %e)
> +        sys.exit(1)

Since you're just printing the exception you might as well omit the
exception handler.  Let Python's default unhandled exception handler
terminate the script instead of duplicating that here.

> +
> +    # Python 2.4 doesn't support 'finally' and 'except' in the same 'try'
> +    # block
> +    try:
> +        try:
> +            test.execute(command, seed)
> +            #Silent exit on user break
> +        except (KeyboardInterrupt, SystemExit):
> +            sys.exit(1)
> +        except:
> +            e = sys.exc_info()[1]
> +            print("FAIL: %s"  %e)
> +            sys.exit(1)

Same thing about unhandled exceptions here.

> +    finally:
> +        test.finish()
> -- 
> 1.8.2.1
> 
>