# 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 os
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--'
[docs]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
[docs] 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
[docs]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]]
[docs] 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)
[docs] 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)
[docs] 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?")
[docs] def old_call(self, name):
"""Get the original caller"""
for arg in self.mocks:
if arg[1] == name:
return arg[2]
return lambda: None
[docs]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]
[docs] 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()
[docs] @classmethod
def cmddir(cls):
"""Returns the location of all the mocked command results"""
from . import TestCase
return os.path.join(TestCase._mockdatadir, 'cmd')
[docs] 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
[docs] def clean_paths(self, data, files):
"""Clean a string of any files or tempdirs"""
try:
for fname in files:
data = data.replace(fname, os.path.basename(fname))
for fdir in self.recorded_tempdirs:
data = data.replace(fdir, '.')
except (UnicodeDecodeError, TypeError):
pass
return data
[docs] 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
[docs] 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
[docs] 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'] = 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'] = 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: {}/{}".format(program, key))
[docs] 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
[docs] @staticmethod
def add_call_file(msg, filename):
"""Add a single file to the given mime message"""
fname = os.path.basename(filename)
with open(filename, "rb") as flh:
part = MIMEApplication(flh.read(), Name=fname)
# After the file is closed
part['Content-Disposition'] = 'attachment'
part['Filename'] = fname
msg.attach(part)
[docs] 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
[docs] 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(), 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
[docs] 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
[docs] 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'))
[docs] 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'))