123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141 |
- #!/usr/bin/env python3
- #
- # Copyright 2018 The Chromium OS Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- #
- """A tool for running Puffin tests in a corpus of deflate compressed files."""
- import argparse
- import filecmp
- import logging
- import os
- import subprocess
- import sys
- import tempfile
- _PUFFHUFF = 'puffhuff'
- _PUFFDIFF = 'puffdiff'
- TESTS = (_PUFFHUFF, _PUFFDIFF)
- class Error(Exception):
- """Puffin general processing error."""
- def ParseArguments(argv):
- """Parses and Validates command line arguments.
- Args:
- argv: command line arguments to parse.
- Returns:
- The arguments list.
- """
- parser = argparse.ArgumentParser()
- parser.add_argument('corpus', metavar='CORPUS',
- help='A corpus directory containing compressed files')
- parser.add_argument('-d', '--disabled_tests', default=(), metavar='',
- nargs='*',
- help=('Space separated list of tests to disable. '
- 'Allowed options include: ' + ', '.join(TESTS)),
- choices=TESTS)
- parser.add_argument('--cache_size', type=int, metavar='SIZE',
- help='The size (in bytes) of the cache for puffpatch '
- 'operations.')
- parser.add_argument('--debug', action='store_true',
- help='Turns on verbosity.')
- # Parse command-line arguments.
- args = parser.parse_args(argv)
- if not os.path.isdir(args.corpus):
- raise Error('Corpus directory {} is non-existent or inaccesible'
- .format(args.corpus))
- return args
- def main(argv):
- """The main function."""
- args = ParseArguments(argv[1:])
- if args.debug:
- logging.getLogger().setLevel(logging.DEBUG)
- # Construct list of appropriate files.
- files = list(filter(os.path.isfile, [os.path.join(args.corpus, f)
- for f in os.listdir(args.corpus)]))
- # For each file in corpus run puffhuff.
- if _PUFFHUFF not in args.disabled_tests:
- for src in files:
- with tempfile.NamedTemporaryFile() as tgt_file:
- operation = 'puffhuff'
- logging.debug('Running %s on %s', operation, src)
- cmd = ['puffin',
- '--operation={}'.format(operation),
- '--src_file={}'.format(src),
- '--dst_file={}'.format(tgt_file.name)]
- if subprocess.call(cmd) != 0:
- raise Error('Puffin failed to do {} command: {}'
- .format(operation, cmd))
- if not filecmp.cmp(src, tgt_file.name):
- raise Error('The generated file {} is not equivalent to the original '
- 'file {} after {} operation.'
- .format(tgt_file.name, src, operation))
- if _PUFFDIFF not in args.disabled_tests:
- # Run puffdiff and puffpatch for each pairs of files in the corpus.
- for src in files:
- for tgt in files:
- with tempfile.NamedTemporaryFile() as patch, \
- tempfile.NamedTemporaryFile() as new_tgt:
- operation = 'puffdiff'
- logging.debug('Running %s on %s (%d) and %s (%d)',
- operation,
- os.path.basename(src), os.stat(src).st_size,
- os.path.basename(tgt), os.stat(tgt).st_size)
- cmd = ['puffin',
- '--operation={}'.format(operation),
- '--src_file={}'.format(src),
- '--dst_file={}'.format(tgt),
- '--patch_file={}'.format(patch.name)]
- # Running the puffdiff operation
- if subprocess.call(cmd) != 0:
- raise Error('Puffin failed to do {} command: {}'
- .format(operation, cmd))
- logging.debug('Patch size is: %d', os.stat(patch.name).st_size)
- operation = 'puffpatch'
- logging.debug('Running %s on src file %s and patch %s',
- operation, os.path.basename(src), patch.name)
- cmd = ['puffin',
- '--operation={}'.format(operation),
- '--src_file={}'.format(src),
- '--dst_file={}'.format(new_tgt.name),
- '--patch_file={}'.format(patch.name)]
- if args.cache_size:
- cmd += ['--cache_size={}'.format(args.cache_size)]
- # Running the puffpatch operation
- if subprocess.call(cmd) != 0:
- raise Error('Puffin failed to do {} command: {}'
- .format(operation, cmd))
- if not filecmp.cmp(tgt, new_tgt.name):
- raise Error('The generated file {} is not equivalent to the '
- 'original file {} after puffpatch operation.'
- .format(new_tgt.name, tgt))
- return 0
- if __name__ == '__main__':
- sys.exit(main(sys.argv))
|