tests/compare.py

Thu, 10 Jul 2008 19:51:53 -0400

author
Brett Smith <brett@brettcsmith.org>
date
Thu, 10 Jul 2008 19:51:53 -0400
branch
trunk
changeset 69
35a2f45cdd3b
parent 54
cd43d2f61162
child 107
1b7450ae4c67
permissions
-rw-r--r--

Count files in the archive and report that in the recursion prompt.

I like this because, for example, if you see that all or most of the files
in the archive are recursive, you can go ahead and decide to recurse right
away.

This invoved making the grep tests a little smarter about handling white
space.

#!/usr/bin/env python
#
# compare.py -- High-level tests for dtrx.
# Copyright (c) 2006, 2007, 2008 Brett Smith <brettcsmith@brettcsmith.org>.
#
# 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 os
import re
import subprocess
import syck
import sys
import tempfile

from sets import Set as set

if os.path.exists('scripts/dtrx') and os.path.exists('tests'):
    os.chdir('tests')
elif os.path.exists('../scripts/dtrx') and os.path.exists('../tests'):
    pass
else:
    print "ERROR: Can't run tests in this directory!"
    sys.exit(2)

X_SCRIPT = os.path.realpath('../scripts/dtrx')
ROOT_DIR = os.path.realpath(os.curdir)
OUTCOMES = ['error', 'failed', 'passed']
TESTSCRIPT_NAME = 'testscript.sh'
SCRIPT_PROLOGUE = """#!/bin/sh
set -e
"""

input_buffer = tempfile.TemporaryFile()
output_buffer = tempfile.TemporaryFile()

class ExtractorTestError(Exception):
    pass


class ExtractorTest(object):
    def __init__(self, **kwargs):
        setattr(self, 'name', kwargs['name'])
        setattr(self, 'options', kwargs.get('options', '-n').split())
        setattr(self, 'filenames', kwargs.get('filenames', '').split())
        for key in ('directory', 'prerun', 'posttest', 'baseline', 'error',
                    'grep', 'antigrep', 'input', 'output', 'cleanup'):
            setattr(self, key, kwargs.get(key, None))
        
    def get_results(self, commands, stdin=None):
        print >>output_buffer, "Output from %s:" % (' '.join(commands),)
        output_buffer.flush()
        status = subprocess.call(commands, stdout=output_buffer,
                                 stderr=output_buffer, stdin=stdin)
        process = subprocess.Popen(['find', '!', '-name', TESTSCRIPT_NAME],
                                   stdout=subprocess.PIPE)
        process.wait()
        output = process.stdout.read(-1)
        process.stdout.close()
        return status, set(output.split('\n'))
        
    def write_script(self, commands):
        script = open(TESTSCRIPT_NAME, 'w')
        script.write("%s%s\n" % (SCRIPT_PROLOGUE, commands))
        script.close()
        subprocess.call(['chmod', 'u+w', TESTSCRIPT_NAME])

    def run_script(self, key):
        commands = getattr(self, key)
        if commands is not None:
            if self.directory:
                directory_hint = '../'
            else:
                directory_hint = ''
            self.write_script(commands)
            subprocess.call(['sh', TESTSCRIPT_NAME, directory_hint])

    def get_shell_results(self):
        self.run_script('prerun')
        self.write_script(self.baseline)
        return self.get_results(['sh', TESTSCRIPT_NAME] + self.filenames)

    def get_extractor_results(self):
        self.run_script('prerun')
        input_buffer.seek(0, 0)
        input_buffer.truncate()
        if self.input:
            input_buffer.write(self.input)
            if not self.input.endswith('\n'):
                input_buffer.write('\n')
            input_buffer.seek(0, 0)
        input_buffer.flush()
        return self.get_results([X_SCRIPT] + self.options + self.filenames,
                                input_buffer)
        
    def get_posttest_result(self):
        if not self.posttest:
            return 0
        self.write_script(self.posttest)
        return subprocess.call(['sh', TESTSCRIPT_NAME])

    def clean(self):
        self.run_script('cleanup')
        if self.directory:
            target = os.path.join(ROOT_DIR, self.directory)
            extra_options = ['!', '-name', TESTSCRIPT_NAME]
        else:
            target = ROOT_DIR
            extra_options = ['(', '(', '-type', 'd',
                             '!', '-name', 'CVS',
                             '!', '-name', '.svn', ')',
                             '-or', '-name', 'test-text',
                             '-or', '-name', 'test-onefile', ')']
        status = subprocess.call(['find', target,
                                  '-mindepth', '1', '-maxdepth', '1'] +
                                 extra_options +
                                 ['-exec', 'rm', '-rf', '{}', ';'])
        if status != 0:
            raise ExtractorTestError("cleanup exited with status code %s" %
                                     (status,))

    def show_status(self, status, message=None):
        raw_status = status.lower()
        if raw_status != 'passed':
            output_buffer.seek(0, 0)
            sys.stdout.write(output_buffer.read(-1))
        if message is None:
            last_part = ''
        else:
            last_part = ': %s' % (message,)
        print "%7s: %s%s" % (status, self.name, last_part)
        return raw_status

    def compare_results(self, actual):
        posttest_result = self.get_posttest_result()
        self.clean()
        status, expected = self.get_shell_results()
        self.clean()
        if expected != actual:
            print >>output_buffer, "Only in baseline results:"
            print >>output_buffer, '\n'.join(expected.difference(actual))
            print >>output_buffer, "Only in actual results:"
            print >>output_buffer, '\n'.join(actual.difference(expected))
            return self.show_status('FAILED')
        elif posttest_result != 0:
            print >>output_buffer, "Posttest gave status code", posttest_result
            return self.show_status('FAILED')
        return self.show_status('Passed')
    
    def have_error_mismatch(self, status):
        if self.error and (status == 0):
            return "dtrx did not return expected error"
        elif (not self.error) and (status != 0):
            return "dtrx returned error code %s" % (status,)
        return None

    def grep_output(self, output):
        if self.grep and (not re.search(self.grep.replace(' ', '\\s+'),
                                        output, re.MULTILINE)):
            return "output did not match %s" % (self.grep)
        elif self.antigrep and re.search(self.antigrep.replace(' ', '\\s+'),
                                         output, re.MULTILINE):
            return "output matched antigrep %s" % (self.antigrep)
        return None

    def check_output(self, output):
        if ((self.output is not None) and
            (self.output.strip() != output.strip())):
            return "output did not match provided text"
        return None

    def check_results(self):
        output_buffer.seek(0, 0)
        output_buffer.truncate()
        self.clean()
        status, actual = self.get_extractor_results()
        output_buffer.seek(0, 0)
        output_buffer.readline()
        output = output_buffer.read(-1)
        problem = (self.have_error_mismatch(status) or
                   self.check_output(output) or self.grep_output(output))
        if problem:
            return self.show_status('FAILED', problem)
        if self.baseline:
            return self.compare_results(actual)
        else:
            self.clean()
            return self.show_status('Passed')

    def run(self):
        if self.directory:
            os.mkdir(self.directory)
            os.chdir(self.directory)
        try:
            result = self.check_results()
        except ExtractorTestError, error:
            result = self.show_status('ERROR', error)
        if self.directory:
            os.chdir(ROOT_DIR)
            subprocess.call(['chmod', '-R', '700', self.directory])
            subprocess.call(['rm', '-rf', self.directory])
        return result


test_db = open('tests.yml')
test_data = syck.load(test_db.read(-1))
test_db.close()
tests = [ExtractorTest(**data) for data in test_data]
for original_data in test_data:
    if (original_data.has_key('directory') or
        (not original_data.has_key('baseline'))):
        continue
    data = original_data.copy()
    data['name'] += ' in ..'
    data['directory'] = 'inside-dir'
    data['filenames'] = ' '.join(['../%s' % filename for filename in
                                  data.get('filenames', '').split()])
    tests.append(ExtractorTest(**data))
results = [test.run() for test in tests]
counts = {}
for outcome in OUTCOMES:
    counts[outcome] = 0
for result in results:
    counts[result] += 1
print " Totals:", ', '.join(["%s %s" % (counts[key], key) for key in OUTCOMES])
input_buffer.close()
output_buffer.close()

mercurial