#!/usr/bin/env python # Copyright 2013 Google Inc. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """ This script will take as an argument either a list of skp files or a set of directories that contains skp files. It will then test each skp file with the `render_pictures` program. If that program either spits out any unexpected output or doesn't return 0, I will flag that skp file as problematic. We then extract all of the embedded images inside the skp and test each one of them against the SkImageDecoder::DecodeFile function. Again, we consider any extraneous output or a bad return value an error. In the event of an error, we retain the image and print out information about the error. The output (on stdout) is formatted as a csv document. A copy of each bad image is left in a directory created by tempfile.mkdtemp(). """ import glob import os import re import shutil import subprocess import sys import tempfile import threading import test_rendering # skia/trunk/tools. reuse FindPathToProgram() USAGE = """ Usage: {command} SKP_FILE [SKP_FILES] {command} SKP_DIR [SKP_DIRS]\n Environment variables: To run multiple worker threads, set NUM_THREADS. To use a different temporary storage location, set TMPDIR. """ def execute_program(args, ignores=None): """ Execute a process and waits for it to complete. Returns all output (stderr and stdout) after (optional) filtering. @param args is passed into subprocess.Popen(). @param ignores (optional) is a list of regular expression strings that will be ignored in the output. @returns a tuple (returncode, output) """ if ignores is None: ignores = [] else: ignores = [re.compile(ignore) for ignore in ignores] proc = subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) output = ''.join( line for line in proc.stdout if not any(bool(ignore.match(line)) for ignore in ignores)) returncode = proc.wait() return (returncode, output) def list_files(paths): """ Accepts a list of directories or filenames on the command line. We do not choose to recurse into directories beyond one level. """ class NotAFileException(Exception): pass for path in paths: for globbedpath in glob.iglob(path): # useful on win32 if os.path.isdir(globbedpath): for filename in os.listdir(globbedpath): newpath = os.path.join(globbedpath, filename) if os.path.isfile(newpath): yield newpath elif os.path.isfile(globbedpath): yield globbedpath else: raise NotAFileException('{} is not a file'.format(globbedpath)) class BadImageFinder(object): def __init__(self, directory=None): self.render_pictures = test_rendering.FindPathToProgram( 'render_pictures') self.test_image_decoder = test_rendering.FindPathToProgram( 'test_image_decoder') assert os.path.isfile(self.render_pictures) assert os.path.isfile(self.test_image_decoder) if directory is None: self.saved_image_dir = tempfile.mkdtemp(prefix='skia_skp_test_') else: assert os.path.isdir(directory) self.saved_image_dir = directory self.bad_image_count = 0 def process_files(self, skp_files): for path in skp_files: self.process_file(path) def process_file(self, skp_file): assert self.saved_image_dir is not None assert os.path.isfile(skp_file) args = [self.render_pictures, '--readPath', skp_file] ignores = ['^process_in', '^deserializ', '^drawing...', '^Non-defaul'] returncode, output = execute_program(args, ignores) if (returncode == 0) and not output: return temp_image_dir = tempfile.mkdtemp(prefix='skia_skp_test___') args = [ self.render_pictures, '--readPath', skp_file, '--writePath', temp_image_dir, '--writeEncodedImages'] subprocess.call(args, stderr=open(os.devnull,'w'), stdout=open(os.devnull,'w')) for image_name in os.listdir(temp_image_dir): image_path = os.path.join(temp_image_dir, image_name) assert(os.path.isfile(image_path)) args = [self.test_image_decoder, image_path] returncode, output = execute_program(args, []) if (returncode == 0) and not output: os.remove(image_path) continue try: shutil.move(image_path, self.saved_image_dir) except (shutil.Error,): # If this happens, don't stop the entire process, # just warn the user. os.remove(image_path) sys.stderr.write('{0} is a repeat.\n'.format(image_name)) self.bad_image_count += 1 if returncode == 2: returncode = 'SkImageDecoder::DecodeFile returns false' elif returncode == 0: returncode = 'extra verbosity' assert output elif returncode == -11: returncode = 'segmentation violation' else: returncode = 'returncode: {}'.format(returncode) output = output.strip().replace('\n',' ').replace('"','\'') suffix = image_name[-3:] output_line = '"{0}","{1}","{2}","{3}","{4}"\n'.format( returncode, suffix, skp_file, image_name, output) sys.stdout.write(output_line) sys.stdout.flush() os.rmdir(temp_image_dir) return def main(main_argv): if not main_argv or main_argv[0] in ['-h', '-?', '-help', '--help']: sys.stderr.write(USAGE.format(command=__file__)) return 1 if 'NUM_THREADS' in os.environ: number_of_threads = int(os.environ['NUM_THREADS']) if number_of_threads < 1: number_of_threads = 1 else: number_of_threads = 1 os.environ['skia_images_png_suppressDecoderWarnings'] = 'true' os.environ['skia_images_jpeg_suppressDecoderWarnings'] = 'true' temp_dir = tempfile.mkdtemp(prefix='skia_skp_test_') sys.stderr.write('Directory for bad images: {}\n'.format(temp_dir)) sys.stdout.write('"Error","Filetype","SKP File","Image File","Output"\n') sys.stdout.flush() finders = [ BadImageFinder(temp_dir) for index in xrange(number_of_threads)] arguments = [[] for index in xrange(number_of_threads)] for index, item in enumerate(list_files(main_argv)): ## split up the given targets among the worker threads arguments[index % number_of_threads].append(item) threads = [ threading.Thread( target=BadImageFinder.process_files, args=(finder,argument)) for finder, argument in zip(finders, arguments)] for thread in threads: thread.start() for thread in threads: thread.join() number = sum(finder.bad_image_count for finder in finders) sys.stderr.write('Number of bad images found: {}\n'.format(number)) return 0 if __name__ == '__main__': exit(main(sys.argv[1:])) # LocalWords: skp stdout csv