diff options
Diffstat (limited to 'share/extensions/inkex/tester/mock.py')
| -rw-r--r-- | share/extensions/inkex/tester/mock.py | 414 |
1 files changed, 414 insertions, 0 deletions
diff --git a/share/extensions/inkex/tester/mock.py b/share/extensions/inkex/tester/mock.py new file mode 100644 index 0000000..2df8791 --- /dev/null +++ b/share/extensions/inkex/tester/mock.py @@ -0,0 +1,414 @@ +# coding=utf-8 +# +# Copyright (C) 2018 Martin Owens +# +# 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 2 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, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110, USA. +# +# pylint: disable=protected-access,too-few-public-methods +""" +Any mocking utilities required by testing. Mocking is when you need the test +to exercise a piece of code, but that code may or does call on something +outside of the target code that either takes too long to run, isn't available +during the test running process or simply shouldn't be running at all. +""" + +import io +import os +import sys +import logging +import hashlib +import tempfile + +from email.mime.application import MIMEApplication +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText +from email.parser import Parser as EmailParser + +import inkex.command + +if False: # pylint: disable=using-constant-test + from typing import List, Tuple, Callable, Any # pylint: disable=unused-import + +FIXED_BOUNDARY = '--CALLDATA--//--CALLDATA--' + +class Capture(object): + """Capture stdout or stderr. Used as `with Capture('stdout') as stream:`""" + def __init__(self, io_name='stdout', swap=True): + self.io_name = io_name + self.original = getattr(sys, io_name) + self.stream = io.StringIO() + self.swap = swap + + def __enter__(self): + # We can't control python2 correctly (unicode vs. bytes-like) but + # we don't need it, so we're ignore python2 as if it doesn't exist. + if self.swap: + setattr(sys, self.io_name, self.stream) + return self.stream + + def __exit__(self, exc, value, traceback): + if exc is not None and self.swap: + # Dump content back to original if there was an error. + self.original.write(self.stream.getvalue()) + setattr(sys, self.io_name, self.original) + +class ManualVerbosity(object): + """Change the verbosity of the test suite manually""" + result = property(lambda self: self.test._current_result) + + def __init__(self, test, okay=True, dots=False): + self.test = test + self.okay = okay + self.dots = dots + + def flip(self, exc_type=None, exc_val=None, exc_tb=None): # pylint: disable=unused-argument + """Swap the stored verbosity with the original""" + self.okay, self.result.showAll = self.result.showAll, self.okay + self.dots, self.result.dots = self.result.dots, self.okay + + __enter__ = flip + __exit__ = flip + + +class MockMixin(object): + """ + Add mocking ability to any test base class, will set up mock on setUp + and remove it on tearDown. + + Mocks are stored in an array attached to the test class (not instance!) which + ensures that mocks can only ever be setUp once and can never be reset over + themselves. (just in case this looks weird at first glance) + + class SomeTest(MockingMixin, TestBase): + mocks = [(sys, 'exit', NoSystemExit("Nope!")] + """ + mocks = [] # type: List[Tuple[Any, str, Any]] + + def setUpMock(self, owner, name, new): # pylint: disable=invalid-name + """Setup the mock here, taking name and function and returning (name, old)""" + old = getattr(owner, name) + if isinstance(new, str): + if hasattr(self, new): + new = getattr(self, new) + if isinstance(new, Exception): + def _error_function(*args2, **kw2): # pylint: disable=unused-argument + raise type(new)(str(new)) + setattr(owner, name, _error_function) + elif new is None or isinstance(new, (str, int, float, list, tuple)): + def _value_function(*args, **kw): # pylint: disable=unused-argument + return new + setattr(owner, name, _value_function) + else: + setattr(owner, name, new) + # When we start, mocks contains length 3 tuples, when we're finished, it contains + # length 4, this stops remocking and reunmocking from taking place. + return (owner, name, old, False) + + def setUp(self): # pylint: disable=invalid-name + """For each mock instruction, set it up and store the return""" + super(MockMixin, self).setUp() + for x, mock in enumerate(self.mocks): + if len(mock) == 4: + logging.error("Mock was already set up, so it wasn't cleared previously!") + continue + self.mocks[x] = self.setUpMock(*mock) + + def tearDown(self): # pylint: disable=invalid-name + """For each returned stored, tear it down and restore mock instruction""" + super(MockMixin, self).tearDown() + try: + for x, (owner, name, old, _) in enumerate(self.mocks): + self.mocks[x] = (owner, name, getattr(owner, name)) + setattr(owner, name, old) + except ValueError: + logging.warning("Was never mocked, did something go wrong?") + + def old_call(self, name): + """Get the original caller""" + for arg in self.mocks: + if arg[1] == name: + return arg[2] + return lambda: None + +class MockCommandMixin(MockMixin): + """ + Replace all the command functions with testable replacements. + + This stops the pipeline and people without the programs, running into problems. + """ + mocks = [ + (inkex.command, '_call', 'mock_call'), + (tempfile, 'mkdtemp', 'record_tempdir'), + ] + recorded_tempdirs = [] # type:List[str] + + def setUp(self): # pylint: disable=invalid-name + super(MockCommandMixin, self).setUp() + # This is a the daftest thing I've ever seen, when in the middle + # of a mock, the 'self' variable magically turns from a FooTest + # into a TestCase, this makes it impossible to find the datadir. + from . import TestCase + TestCase._mockdatadir = self.datadir() + + @classmethod + def cmddir(cls): + """Returns the location of all the mocked command results""" + from . import TestCase + return os.path.join(TestCase._mockdatadir, 'cmd') + + def record_tempdir(self, *args, **kwargs): + """Record any attempts to make tempdirs""" + newdir = self.old_call('mkdtemp')(*args, **kwargs) + self.recorded_tempdirs.append(newdir) + return newdir + + def clean_paths(self, data, files): + """Clean a string of any files or tempdirs""" + try: + for fdir in self.recorded_tempdirs: + data = data.replace(fdir, '.') + files = [fname.replace(fdir, '.') for fname in files] + for fname in files: + data = data.replace(fname, os.path.basename(fname)) + except (UnicodeDecodeError, TypeError): + pass + return data + + def get_all_tempfiles(self): + """Returns a set() of all files currently in any of the tempdirs""" + ret = set([]) + for fdir in self.recorded_tempdirs: + if not os.path.isdir(fdir): + continue + for fname in os.listdir(fdir): + if fname in ('.', '..'): + continue + path = os.path.join(fdir, fname) + # We store the modified time so if a program modifies + # the input file in-place, it will look different. + ret.add(path + ';{}'.format(os.path.getmtime(path))) + + return ret + + def ignore_command_mock(self, program, arglst): + """Return true if the mock is ignored""" + if self and program and arglst: + return os.environ.get('NO_MOCK_COMMANDS') + return False + + def mock_call(self, program, *args, **kwargs): + """ + Replacement for the inkex.command.call() function, instead of calling + an external program, will compile all arguments into a hash and use the + hash to find a command result. + """ + # Remove stdin first because it needs to NOT be in the Arguments list. + stdin = kwargs.pop('stdin', None) + args = list(args) + + # We use email + msg = MIMEMultipart(boundary=FIXED_BOUNDARY) + msg['Program'] = self.get_program_name(program) + + # Gather any output files and add any input files to msg, args and kwargs + # may be modified to strip out filename directories (which change) + inputs, outputs = self.add_call_files(msg, args, kwargs) + + arglst = inkex.command.to_args(program, *args, **kwargs)[1:] + arglst.sort() + argstr = ' '.join(arglst) + argstr = self.clean_paths(argstr, inputs + outputs) + msg['Arguments'] = argstr.strip() + + if stdin is not None: + # The stdin is counted as the msg body + cleanin = self.clean_paths(stdin, inputs + outputs) + msg.attach(MIMEText(cleanin, 'plain', 'utf-8')) + + keystr = msg.as_string() + # There is a difference between python2 and python3 output + keystr = keystr.replace('\n\n', '\n') + keystr = keystr.replace('\n ', ' ') + if 'verb' in keystr: + # Verbs seperated by colons cause diff in py2/3 + keystr = keystr.replace('; ', ';') + # Generate a unique key for this call based on _all_ it's inputs + key = hashlib.md5(keystr.encode('utf-8')).hexdigest() + + if self.ignore_command_mock(program, arglst): + # Call original code. This is so programmers can run the test suite + # against the external programs too, to see how their fair. + if stdin is not None: + kwargs['stdin'] = stdin + + before = self.get_all_tempfiles() + stdout = self.old_call('_call')(program, *args, **kwargs) + outputs += list(self.get_all_tempfiles() - before) + # Remove the modified time from the call + outputs = [out.rsplit(';', 1)[0] for out in outputs] + + # After the program has run, we collect any file outputs and store + # them, then store any stdout or stderr created during the run. + # A developer can then use this to build new test cases. + reply = MIMEMultipart(boundary=FIXED_BOUNDARY) + reply['Program'] = self.get_program_name(program) + reply['Arguments'] = argstr + self.save_call(program, key, stdout, outputs, reply) + self.save_key(program, key, keystr, 'key') + return stdout + + try: + return self.load_call(program, key, outputs) + except IOError: + self.save_key(program, key, keystr, 'bad-key') + raise IOError("Problem loading call: {}/{} use the environment variable "\ + "NO_MOCK_COMMANDS=1 to call out to the external program and generate "\ + "the mock call file.".format(program, key)) + + def add_call_files(self, msg, args, kwargs): + """ + Gather all files, adding input files to the msg (for hashing) and + output files to the returned files list (for outputting in debug) + """ + # Gather all possible string arguments together. + loargs = sorted(kwargs.items(), key=lambda i: i[0]) + values = [] + for arg in args: + if isinstance(arg, (tuple, list)): + loargs.append(arg) + else: + values.append(str(arg)) + + for (_, value) in loargs: + if isinstance(value, (tuple, list)): + for val in value: + if val is not True: + values.append(str(val)) + elif value is not True: + values.append(str(value)) + + # See if any of the strings could be filenames, either going to be + # or are existing files on the disk. + files = [[], []] + for value in values: + if os.path.isfile(value): # Input file + files[0].append(value) + self.add_call_file(msg, value) + elif os.path.isdir(os.path.dirname(value)): # Output file + files[1].append(value) + return files + + def add_call_file(self, msg, filename): + """Add a single file to the given mime message""" + fname = os.path.basename(filename) + with open(filename, "rb") as fhl: + if filename.endswith('.svg'): + value = self.clean_paths(fhl.read().decode('utf8'), []) + else: + value = fhl.read() + part = MIMEApplication(value, Name=fname) + # After the file is closed + part['Content-Disposition'] = 'attachment' + part['Filename'] = fname + msg.attach(part) + + def get_call_filename(self, program, key, create=False): + """ + Get the filename for the call testing information. + """ + path = self.get_call_path(program, create=create) + fname = os.path.join(path, key + '.msg') + if not create and not os.path.isfile(fname): + raise IOError("Attempted to find call test data {}".format(key)) + return fname + + def get_program_name(self, program): + """Takes a program and returns a program name""" + if program == inkex.command.INKSCAPE_EXECUTABLE_NAME: + return 'inkscape' + return program + + def get_call_path(self, program, create=True): + """Get where this program would store it's test data""" + command_dir = os.path.join(self.cmddir(), self.get_program_name(program)) + if not os.path.isdir(command_dir): + if create: + os.makedirs(command_dir) + else: + raise IOError("A test is attempting to use an external program in a test:"\ + " {}; but there is not a command data directory which should"\ + " contain the results of the command here: {}"\ + .format(program, command_dir)) + return command_dir + + def load_call(self, program, key, files): + """ + Load the given call + """ + fname = self.get_call_filename(program, key, create=False) + with open(fname, 'rb') as fhl: + msg = EmailParser().parsestr(fhl.read().decode('utf-8')) + + stdout = None + for part in msg.walk(): + if 'attachment' in part.get("Content-Disposition", ''): + base_name = part['Filename'] + for out_file in files: + if out_file.endswith(base_name): + with open(out_file, 'wb') as fhl: + fhl.write(part.get_payload(decode=True)) + part = None + if part is not None: + # Was not caught by any normal outputs, so we will + # save the file to EVERY tempdir in the hopes of + # hitting on of them. + for fdir in self.recorded_tempdirs: + if os.path.isdir(fdir): + with open(os.path.join(fdir, base_name), 'wb') as fhl: + fhl.write(part.get_payload(decode=True)) + elif part.get_content_type() == "text/plain": + stdout = part.get_payload(decode=True) + + return stdout + + def save_call(self, program, key, stdout, files, msg, ext='output'): # pylint: disable=too-many-arguments + """ + Saves the results from the call into a debug output file, the resulting files + should be a Mime msg file format with each attachment being one of the input + files as well as any stdin and arguments used in the call. + """ + if stdout is not None and stdout.strip(): + # The stdout is counted as the msg body here + msg.attach(MIMEText(stdout.decode('utf-8'), 'plain', 'utf-8')) + + for fname in set(files): + if os.path.isfile(fname): + #print("SAVING FILE INTO MSG: {}".format(fname)) + self.add_call_file(msg, fname) + else: + part = MIMEText("Missing File", 'plain', 'utf-8') + part.add_header('Filename', os.path.basename(fname)) + msg.attach(part) + + fname = self.get_call_filename(program, key, create=True) + '.' + ext + with open(fname, 'wb') as fhl: + fhl.write(msg.as_string().encode('utf-8')) + + def save_key(self, program, key, keystr, ext='key'): + """Save the key file if we are debugging the key data""" + if os.environ.get('DEBUG_KEY'): + fname = self.get_call_filename(program, key, create=True) + '.' + ext + with open(fname, 'wb') as fhl: + fhl.write(keystr.encode('utf-8')) |