#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# compare.py -- High-level tests for dtrx.
# Copyright © 2006-2009 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 yaml
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',
'input', 'output', 'cleanup'):
setattr(self, key, kwargs.get(key, None))
for key in ('grep', 'antigrep'):
value = kwargs.get(key, [])
if isinstance(value, str):
value = [value]
setattr(self, key, value)
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):
for pattern in self.grep:
if not re.search(pattern.replace(' ', '\\s+'), output,
re.MULTILINE):
return "output did not match %s" % (pattern)
for pattern in self.antigrep:
if re.search(pattern.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 = yaml.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()