# Copyright 2002-2005 Vladimir Prus. # Copyright 2002-2003 Dave Abrahams. # Copyright 2006 Rene Rivera. # Distributed under the Boost Software License, Version 1.0. # (See accompanying file LICENSE_1_0.txt or copy at # http://www.boost.org/LICENSE_1_0.txt) import TestCmd import copy import fnmatch import glob import math import os import os.path import re import shutil import StringIO import subprocess import sys import tempfile import time import traceback import tree import types from xml.sax.saxutils import escape class TestEnvironmentError(Exception): pass annotations = [] def print_annotation(name, value, xml): """Writes some named bits of information about the current test run.""" if xml: print escape(name) + " {{{" print escape(value) print "}}}" else: print name + " {{{" print value print "}}}" def flush_annotations(xml=0): global annotations for ann in annotations: print_annotation(ann[0], ann[1], xml) annotations = [] def clear_annotations(): global annotations annotations = [] defer_annotations = 0 def set_defer_annotations(n): global defer_annotations defer_annotations = n def annotate_stack_trace(tb=None): if tb: trace = TestCmd.caller(traceback.extract_tb(tb), 0) else: trace = TestCmd.caller(traceback.extract_stack(), 1) annotation("stacktrace", trace) def annotation(name, value): """Records an annotation about the test run.""" annotations.append((name, value)) if not defer_annotations: flush_annotations() def get_toolset(): toolset = None for arg in sys.argv[1:]: if not arg.startswith("-"): toolset = arg return toolset or "gcc" # Detect the host OS. cygwin = hasattr(os, "uname") and os.uname()[0].lower().startswith("cygwin") windows = cygwin or os.environ.get("OS", "").lower().startswith("windows") def prepare_prefixes_and_suffixes(toolset): prepare_suffix_map(toolset) prepare_library_prefix(toolset) def prepare_suffix_map(toolset): """ Set up suffix translation performed by the Boost Build testing framework to accomodate different toolsets generating targets of the same type using different filename extensions (suffixes). """ global suffixes suffixes = {} if windows: if toolset == "gcc": suffixes[".lib"] = ".a" # mingw static libs use suffix ".a". suffixes[".obj"] = ".o" if cygwin: suffixes[".implib"] = ".lib.a" else: suffixes[".implib"] = ".lib" else: suffixes[".exe"] = "" suffixes[".dll"] = ".so" suffixes[".lib"] = ".a" suffixes[".obj"] = ".o" suffixes[".implib"] = ".no_implib_files_on_this_platform" if hasattr(os, "uname") and os.uname()[0] == "Darwin": suffixes[".dll"] = ".dylib" def prepare_library_prefix(toolset): """ Setup whether Boost Build is expected to automatically prepend prefixes to its built library targets. """ global lib_prefix lib_prefix = "lib" global dll_prefix if cygwin: dll_prefix = "cyg" elif windows and toolset != "gcc": dll_prefix = None else: dll_prefix = "lib" def re_remove(sequence, regex): me = re.compile(regex) result = filter(lambda x: me.match(x), sequence) if not result: raise ValueError() for r in result: sequence.remove(r) def glob_remove(sequence, pattern): result = fnmatch.filter(sequence, pattern) if not result: raise ValueError() for r in result: sequence.remove(r) class Tester(TestCmd.TestCmd): """Main tester class for Boost Build. Optional arguments: `arguments` - Arguments passed to the run executable. `executable` - Name of the executable to invoke. `match` - Function to use for compating actual and expected file contents. `boost_build_path` - Boost build path to be passed to the run executable. `translate_suffixes` - Whether to update suffixes on the the file names passed from the test script so they match those actually created by the current toolset. For example, static library files are specified by using the .lib suffix but when the "gcc" toolset is used it actually creates them using the .a suffix. `pass_toolset` - Whether the test system should pass the specified toolset to the run executable. `use_test_config` - Whether the test system should tell the run executable to read in the test_config.jam configuration file. `ignore_toolset_requirements` - Whether the test system should tell the run executable to ignore toolset requirements. `workdir` - Absolute directory where the test will be run from. `pass_d0` - If set, when tests are not explicitly run in verbose mode, they are run as silent (-d0 & --quiet Boost Jam options). Optional arguments inherited from the base class: `description` - Test description string displayed in case of a failed test. `subdir` - List of subdirectories to automatically create under the working directory. Each subdirectory needs to be specified separately, parent coming before its child. `verbose` - Flag that may be used to enable more verbose test system output. Note that it does not also enable more verbose build system output like the --verbose command line option does. """ def __init__(self, arguments=None, executable="bjam", match=TestCmd.match_exact, boost_build_path=None, translate_suffixes=True, pass_toolset=True, use_test_config=True, ignore_toolset_requirements=True, workdir="", pass_d0=True, **keywords): assert arguments.__class__ is not str self.original_workdir = os.getcwd() if workdir and not os.path.isabs(workdir): raise ("Parameter workdir <%s> must point to an absolute " "directory: " % workdir) self.last_build_timestamp = 0 self.translate_suffixes = translate_suffixes self.use_test_config = use_test_config self.toolset = get_toolset() self.pass_toolset = pass_toolset self.ignore_toolset_requirements = ignore_toolset_requirements prepare_prefixes_and_suffixes(pass_toolset and self.toolset or "gcc") use_default_bjam = "--default-bjam" in sys.argv if not use_default_bjam: jam_build_dir = "" if os.name == "nt": jam_build_dir = "bin.ntx86" elif (os.name == "posix") and os.__dict__.has_key("uname"): if os.uname()[0].lower().startswith("cygwin"): jam_build_dir = "bin.cygwinx86" if ("TMP" in os.environ and os.environ["TMP"].find("~") != -1): print("Setting $TMP to /tmp to get around problem " "with short path names") os.environ["TMP"] = "/tmp" elif os.uname()[0] == "Linux": cpu = os.uname()[4] if re.match("i.86", cpu): jam_build_dir = "bin.linuxx86" else: jam_build_dir = "bin.linux" + os.uname()[4] elif os.uname()[0] == "SunOS": jam_build_dir = "bin.solaris" elif os.uname()[0] == "Darwin": if os.uname()[4] == "i386": jam_build_dir = "bin.macosxx86" elif os.uname()[4] == "x86_64": jam_build_dir = "bin.macosxx86_64" else: jam_build_dir = "bin.macosxppc" elif os.uname()[0] == "AIX": jam_build_dir = "bin.aix" elif os.uname()[0] == "IRIX64": jam_build_dir = "bin.irix" elif os.uname()[0] == "FreeBSD": jam_build_dir = "bin.freebsd" elif os.uname()[0] == "OSF1": jam_build_dir = "bin.osf" else: raise ("Do not know directory where Jam is built for this " "system: %s/%s" % (os.name, os.uname()[0])) else: raise ("Do not know directory where Jam is built for this " "system: %s" % os.name) # Find where jam_src is located. Try for the debug version if it is # lying around. dirs = [os.path.join("..", "src", "engine", jam_build_dir + ".debug"), os.path.join("..", "src", "engine", jam_build_dir)] for d in dirs: if os.path.exists(d): jam_build_dir = d break else: print("Cannot find built Boost.Jam") sys.exit(1) verbosity = ["-d0", "--quiet"] if not pass_d0: verbosity = [] if "--verbose" in sys.argv: keywords["verbose"] = True verbosity = ["-d+2"] if boost_build_path is None: boost_build_path = self.original_workdir + "/.." program_list = [] if use_default_bjam: program_list.append(executable) else: program_list.append(os.path.join(jam_build_dir, executable)) program_list.append('-sBOOST_BUILD_PATH="' + boost_build_path + '"') if verbosity: program_list += verbosity if arguments: program_list += arguments TestCmd.TestCmd.__init__(self, program=program_list, match=match, workdir=workdir, inpath=use_default_bjam, **keywords) os.chdir(self.workdir) def cleanup(self): try: TestCmd.TestCmd.cleanup(self) os.chdir(self.original_workdir) except AttributeError: # When this is called during TestCmd.TestCmd.__del__ we can have # both 'TestCmd' and 'os' unavailable in our scope. Do nothing in # this case. pass # # Methods that change the working directory's content. # def set_tree(self, tree_location): # It is not possible to remove the current directory. d = os.getcwd() os.chdir(os.path.dirname(self.workdir)) shutil.rmtree(self.workdir, ignore_errors=False) if not os.path.isabs(tree_location): tree_location = os.path.join(self.original_workdir, tree_location) shutil.copytree(tree_location, self.workdir) os.chdir(d) def make_writable(unused, dir, entries): for e in entries: name = os.path.join(dir, e) os.chmod(name, os.stat(name).st_mode | 0222) os.path.walk(".", make_writable, None) def write(self, file, content, wait=True): nfile = self.native_file_name(file) self.__makedirs(os.path.dirname(nfile), wait) f = open(nfile, "wb") try: f.write(content) finally: f.close() self.__ensure_newer_than_last_build(nfile) def copy(self, src, dst): try: self.write(dst, self.read(src, 1)) except: self.fail_test(1) def copy_preserving_timestamp(self, src, dst): src_name = self.native_file_name(src) dst_name = self.native_file_name(dst) stats = os.stat(src_name) self.write(dst, self.read(src, 1)) os.utime(dst_name, (stats.st_atime, stats.st_mtime)) def touch(self, names, wait=True): if names.__class__ is str: names = [names] for name in names: path = self.native_file_name(name) if wait: self.__ensure_newer_than_last_build(path) else: os.utime(path, None) def rm(self, names): if not type(names) == types.ListType: names = [names] if names == ["."]: # If we are deleting the entire workspace, there is no need to wait # for a clock tick. self.last_build_timestamp = 0 # Avoid attempts to remove the current directory. os.chdir(self.original_workdir) for name in names: n = glob.glob(self.native_file_name(name)) if n: n = n[0] if not n: n = self.glob_file(name.replace("$toolset", self.toolset + "*") ) if n: if os.path.isdir(n): shutil.rmtree(n, ignore_errors=False) else: os.unlink(n) # Create working dir root again in case we removed it. if not os.path.exists(self.workdir): os.mkdir(self.workdir) os.chdir(self.workdir) def expand_toolset(self, name): """ Expands $toolset placeholder in the given file to the name of the toolset currently being tested. """ self.write(name, self.read(name).replace("$toolset", self.toolset)) def dump_stdio(self): annotation("STDOUT", self.stdout()) annotation("STDERR", self.stderr()) def run_build_system(self, extra_args=None, subdir="", stdout=None, stderr="", status=0, match=None, pass_toolset=None, use_test_config=None, ignore_toolset_requirements=None, expected_duration=None, **kw): assert extra_args.__class__ is not str if os.path.isabs(subdir): print("You must pass a relative directory to subdir <%s>." % subdir ) return self.previous_tree, dummy = tree.build_tree(self.workdir) if match is None: match = self.match if pass_toolset is None: pass_toolset = self.pass_toolset if use_test_config is None: use_test_config = self.use_test_config if ignore_toolset_requirements is None: ignore_toolset_requirements = self.ignore_toolset_requirements try: kw["program"] = [] kw["program"] += self.program if extra_args: kw["program"] += extra_args if pass_toolset: kw["program"].append("toolset=" + self.toolset) if use_test_config: kw["program"].append('--test-config="%s"' % os.path.join( self.original_workdir, "test-config.jam")) if ignore_toolset_requirements: kw["program"].append("--ignore-toolset-requirements") if "--python" in sys.argv: # -z disables Python optimization mode. # this enables type checking (all assert # and if __debug__ statements). kw["program"].extend(["--python", "-z"]) if "--stacktrace" in sys.argv: kw["program"].append("--stacktrace") kw["chdir"] = subdir self.last_program_invocation = kw["program"] build_time_start = time.time() apply(TestCmd.TestCmd.run, [self], kw) build_time_finish = time.time() except: self.dump_stdio() raise old_last_build_timestamp = self.last_build_timestamp self.tree, self.last_build_timestamp = tree.build_tree(self.workdir) self.difference = tree.tree_difference(self.previous_tree, self.tree) if self.difference.empty(): # If nothing has been changed by this build and sufficient time has # passed since the last build that actually changed something, # there is no need to wait for touched or newly created files to # start getting newer timestamps than the currently existing ones. self.last_build_timestamp = old_last_build_timestamp self.difference.ignore_directories() self.unexpected_difference = copy.deepcopy(self.difference) if (status and self.status) is not None and self.status != status: expect = "" if status != 0: expect = " (expected %d)" % status annotation("failure", '"%s" returned %d%s' % (kw["program"], self.status, expect)) annotation("reason", "unexpected status returned by bjam") self.fail_test(1) if stdout is not None and not match(self.stdout(), stdout): annotation("failure", "Unexpected stdout") annotation("Expected STDOUT", stdout) annotation("Actual STDOUT", self.stdout()) stderr = self.stderr() if stderr: annotation("STDERR", stderr) self.maybe_do_diff(self.stdout(), stdout) self.fail_test(1, dump_stdio=False) # Intel tends to produce some messages to stderr which make tests fail. intel_workaround = re.compile("^xi(link|lib): executing.*\n", re.M) actual_stderr = re.sub(intel_workaround, "", self.stderr()) if stderr is not None and not match(actual_stderr, stderr): annotation("failure", "Unexpected stderr") annotation("Expected STDERR", stderr) annotation("Actual STDERR", self.stderr()) annotation("STDOUT", self.stdout()) self.maybe_do_diff(actual_stderr, stderr) self.fail_test(1, dump_stdio=False) if expected_duration is not None: actual_duration = build_time_finish - build_time_start if actual_duration > expected_duration: print("Test run lasted %f seconds while it was expected to " "finish in under %f seconds." % (actual_duration, expected_duration)) self.fail_test(1, dump_stdio=False) def glob_file(self, name): result = None if hasattr(self, "difference"): for f in (self.difference.added_files + self.difference.modified_files + self.difference.touched_files): if fnmatch.fnmatch(f, name): result = self.native_file_name(f) break if not result: result = glob.glob(self.native_file_name(name)) if result: result = result[0] return result def read(self, name, binary=False): try: if self.toolset: name = name.replace("$toolset", self.toolset + "*") name = self.glob_file(name) openMode = "r" if binary: openMode += "b" else: openMode += "U" f = open(name, openMode) result = f.read() f.close() return result except: annotation("failure", "Could not open '%s'" % name) self.fail_test(1) return "" def read_and_strip(self, name): if not self.glob_file(name): return "" f = open(self.glob_file(name), "rb") lines = f.readlines() f.close() result = "\n".join(x.rstrip() for x in lines) if lines and lines[-1][-1] != "\n": return result + "\n" return result def fail_test(self, condition, dump_difference=True, dump_stdio=True, dump_stack=True): if not condition: return if dump_difference and hasattr(self, "difference"): f = StringIO.StringIO() self.difference.pprint(f) annotation("changes caused by the last build command", f.getvalue()) if dump_stdio: self.dump_stdio() if "--preserve" in sys.argv: print print "*** Copying the state of working dir into 'failed_test' ***" print path = os.path.join(self.original_workdir, "failed_test") if os.path.isdir(path): shutil.rmtree(path, ignore_errors=False) elif os.path.exists(path): raise "Path " + path + " already exists and is not a directory" shutil.copytree(self.workdir, path) print "The failed command was:" print " ".join(self.last_program_invocation) if dump_stack: annotate_stack_trace() sys.exit(1) # A number of methods below check expectations with actual difference # between directory trees before and after a build. All the 'expect*' # methods require exact names to be passed. All the 'ignore*' methods allow # wildcards. # All names can be either a string or a list of strings. def expect_addition(self, names): for name in self.adjust_names(names): try: glob_remove(self.unexpected_difference.added_files, name) except: annotation("failure", "File %s not added as expected" % name) self.fail_test(1) def ignore_addition(self, wildcard): self.__ignore_elements(self.unexpected_difference.added_files, wildcard) def expect_removal(self, names): for name in self.adjust_names(names): try: glob_remove(self.unexpected_difference.removed_files, name) except: annotation("failure", "File %s not removed as expected" % name) self.fail_test(1) def ignore_removal(self, wildcard): self.__ignore_elements(self.unexpected_difference.removed_files, wildcard) def expect_modification(self, names): for name in self.adjust_names(names): try: glob_remove(self.unexpected_difference.modified_files, name) except: annotation("failure", "File %s not modified as expected" % name) self.fail_test(1) def ignore_modification(self, wildcard): self.__ignore_elements(self.unexpected_difference.modified_files, wildcard) def expect_touch(self, names): d = self.unexpected_difference for name in self.adjust_names(names): # We need to check both touched and modified files. The reason is # that: # (1) Windows binaries such as obj, exe or dll files have slight # differences even with identical inputs due to Windows PE # format headers containing an internal timestamp. # (2) Intel's compiler for Linux has the same behaviour. filesets = [d.modified_files, d.touched_files] while filesets: try: glob_remove(filesets[-1], name) break except ValueError: filesets.pop() if not filesets: annotation("failure", "File %s not touched as expected" % name) self.fail_test(1) def ignore_touch(self, wildcard): self.__ignore_elements(self.unexpected_difference.touched_files, wildcard) def ignore(self, wildcard): self.ignore_addition(wildcard) self.ignore_removal(wildcard) self.ignore_modification(wildcard) self.ignore_touch(wildcard) def expect_nothing(self, names): for name in self.adjust_names(names): if name in self.difference.added_files: annotation("failure", "File %s added, but no action was expected" % name) self.fail_test(1) if name in self.difference.removed_files: annotation("failure", "File %s removed, but no action was expected" % name) self.fail_test(1) pass if name in self.difference.modified_files: annotation("failure", "File %s modified, but no action was expected" % name) self.fail_test(1) if name in self.difference.touched_files: annotation("failure", "File %s touched, but no action was expected" % name) self.fail_test(1) def expect_nothing_more(self): # Not totally sure about this change, but I do not see a good # alternative. if windows: self.ignore("*.ilk") # MSVC incremental linking files. self.ignore("*.pdb") # MSVC program database files. self.ignore("*.rsp") # Response files. self.ignore("*.tds") # Borland debug symbols. self.ignore("*.manifest") # MSVC DLL manifests. # Debug builds of bjam built with gcc produce this profiling data. self.ignore("gmon.out") self.ignore("*/gmon.out") # Boost Build's 'configure' functionality (unfinished at the time) # produces this file. self.ignore("bin/config.log") self.ignore("bin/project-cache.jam") # Compiled Python files created when running Python based Boost Build. self.ignore("*.pyc") if not self.unexpected_difference.empty(): annotation("failure", "Unexpected changes found") output = StringIO.StringIO() self.unexpected_difference.pprint(output) annotation("unexpected changes", output.getvalue()) self.fail_test(1) def expect_output_lines(self, lines, expected=True): self.__expect_lines(self.stdout(), lines, expected) def expect_content_lines(self, filename, line, expected=True): self.__expect_lines(self.__read_file(filename), line, expected) def expect_content(self, name, content, exact=False): actual = self.__read_file(name, exact) content = content.replace("$toolset", self.toolset + "*") matched = False if exact: matched = fnmatch.fnmatch(actual, content) else: def sorted_(x): x.sort() return x actual_ = map(lambda x: sorted_(x.split()), actual.splitlines()) content_ = map(lambda x: sorted_(x.split()), content.splitlines()) if len(actual_) == len(content_): matched = map( lambda x, y: map(lambda n, p: fnmatch.fnmatch(n, p), x, y), actual_, content_) matched = reduce( lambda x, y: x and reduce( lambda a, b: a and b, y), matched) if not matched: print "Expected:\n" print content print "Got:\n" print actual self.fail_test(1) def maybe_do_diff(self, actual, expected): if os.environ.get("DO_DIFF"): e = tempfile.mktemp("expected") a = tempfile.mktemp("actual") f = open(e, "w") f.write(expected) f.close() f = open(a, "w") f.write(actual) f.close() print("DIFFERENCE") # Current diff should return 1 to indicate 'different input files' # but some older diff versions may return 0 and depending on the # exact Python/OS platform version, os.system() call may gobble up # the external process's return code and return 0 itself. if os.system('diff -u "%s" "%s"' % (e, a)) not in [0, 1]: print('Unable to compute difference: diff -u "%s" "%s"' % (e, a )) os.unlink(e) os.unlink(a) else: print("Set environmental variable 'DO_DIFF' to examine the " "difference.") # Internal methods. def adjust_lib_name(self, name): global lib_prefix global dll_prefix result = name pos = name.rfind(".") if pos != -1: suffix = name[pos:] if suffix == ".lib": (head, tail) = os.path.split(name) if lib_prefix: tail = lib_prefix + tail result = os.path.join(head, tail) elif suffix == ".dll": (head, tail) = os.path.split(name) if dll_prefix: tail = dll_prefix + tail result = os.path.join(head, tail) # If we want to use this name in a Jamfile, we better convert \ to /, # as otherwise we would have to quote \. result = result.replace("\\", "/") return result def adjust_suffix(self, name): if not self.translate_suffixes: return name pos = name.rfind(".") if pos == -1: return name suffix = name[pos:] return name[:pos] + suffixes.get(suffix, suffix) # Acceps either a string or a list of strings and returns a list of # strings. Adjusts suffixes on all names. def adjust_names(self, names): if names.__class__ is str: names = [names] r = map(self.adjust_lib_name, names) r = map(self.adjust_suffix, r) r = map(lambda x, t=self.toolset: x.replace("$toolset", t + "*"), r) return r def native_file_name(self, name): name = self.adjust_names(name)[0] return os.path.normpath(os.path.join(self.workdir, *name.split("/"))) def wait_for_time_change(self, path, touch): """ Wait for newly assigned file system modification timestamps for the given path to become large enough for the timestamp difference to be correctly recognized by both this Python based testing framework and the Boost Jam executable being tested. May optionally touch the given path to set its modification timestamp to the new value. """ self.__wait_for_time_change(path, touch, last_build_time=False) def __build_timestamp_resolution(self): """ Returns the minimum path modification timestamp resolution supported by the used Boost Jam executable. """ dir = tempfile.mkdtemp("bjam_version_info") try: jam_script = "timestamp_resolution.jam" f = open(os.path.join(dir, jam_script), "w") try: f.write("EXIT $(JAM_TIMESTAMP_RESOLUTION) : 0 ;") finally: f.close() p = subprocess.Popen([self.program[0], "-d0", "-f%s" % jam_script], stdout=subprocess.PIPE, cwd=dir, universal_newlines=True) out, err = p.communicate() finally: shutil.rmtree(dir, ignore_errors=False) if p.returncode != 0: raise TestEnvironmentError("Unexpected return code (%s) when " "detecting Boost Jam's minimum supported path modification " "timestamp resolution version information." % p.returncode) if err: raise TestEnvironmentError("Unexpected error output (%s) when " "detecting Boost Jam's minimum supported path modification " "timestamp resolution version information." % err) r = re.match("([0-9]{2}):([0-9]{2}):([0-9]{2}\\.[0-9]{9})$", out) if not r: # Older Boost Jam versions did not report their minimum supported # path modification timestamp resolution and did not actually # support path modification timestamp resolutions finer than 1 # second. # TODO: Phase this support out to avoid such fallback code from # possibly covering up other problems. return 1 if r.group(1) != "00" or r.group(2) != "00": # hours, minutes raise TestEnvironmentError("Boost Jam with too coarse minimum " "supported path modification timestamp resolution (%s:%s:%s)." % (r.group(1), r.group(2), r.group(3))) return float(r.group(3)) # seconds.nanoseconds def __ensure_newer_than_last_build(self, path): """ Updates the given path's modification timestamp after waiting for the newly assigned file system modification timestamp to become large enough for the timestamp difference between it and the last build timestamp to be correctly recognized by both this Python based testing framework and the Boost Jam executable being tested. Does nothing if there is no 'last build' information available. """ if self.last_build_timestamp: self.__wait_for_time_change(path, touch=True, last_build_time=True) def __expect_lines(self, data, lines, expected): """ Checks whether the given data contains the given lines. Data may be specified as a single string containing text lines separated by newline characters. Lines may be specified in any of the following forms: * Single string containing text lines separated by newlines - the given lines are searched for in the given data without any extra data lines between them. * Container of strings containing text lines separated by newlines - the given lines are searched for in the given data with extra data lines allowed between lines belonging to different strings. * Container of strings containing text lines separated by newlines and containers containing strings - the same as above with the internal containers containing strings being interpreted as if all their content was joined together into a single string separated by newlines. A newline at the end of any multi-line lines string is interpreted as an expected extra trailig empty line. """ # str.splitlines() trims at most one trailing newline while we want the # trailing newline to indicate that there should be an extra empty line # at the end. splitlines = lambda x : (x + "\n").splitlines() if data is None: data = [] elif data.__class__ is str: data = splitlines(data) if lines.__class__ is str: lines = [splitlines(lines)] else: expanded = [] for x in lines: if x.__class__ is str: x = splitlines(x) expanded.append(x) lines = expanded if _contains_lines(data, lines) != bool(expected): output = [] if expected: output = ["Did not find expected lines:"] else: output = ["Found unexpected lines:"] first = True for line_sequence in lines: if line_sequence: if first: first = False else: output.append("...") output.extend(" > " + line for line in line_sequence) output.append("in output:") output.extend(" > " + line for line in data) annotation("failure", "\n".join(output)) self.fail_test(1) def __ignore_elements(self, list, wildcard): """Removes in-place 'list' elements matching the given 'wildcard'.""" list[:] = filter(lambda x, w=wildcard: not fnmatch.fnmatch(x, w), list) def __makedirs(self, path, wait): """ Creates a folder with the given path, together with any missing parent folders. If WAIT is set, makes sure any newly created folders have modification timestamps newer than the ones left behind by the last build run. """ try: if wait: stack = [] while path and path not in stack and not os.path.isdir(path): stack.append(path) path = os.path.dirname(path) while stack: path = stack.pop() os.mkdir(path) self.__ensure_newer_than_last_build(path) else: os.makedirs(path) except Exception: pass def __python_timestamp_resolution(self, path, minimum_resolution): """ Returns the modification timestamp resolution for the given path supported by the used Python interpreter/OS/filesystem combination. Will not check for resolutions less than the given minimum value. Will change the path's modification timestamp in the process. Return values: 0 - nanosecond resolution supported positive decimal - timestamp resolution in seconds """ # Note on Python's floating point timestamp support: # Python interpreter versions prior to Python 2.3 did not support # floating point timestamps. Versions 2.3 through 3.3 may or may not # support it depending on the configuration (may be toggled by calling # os.stat_float_times(True/False) at program startup, disabled by # default prior to Python 2.5 and enabled by default since). Python 3.3 # deprecated this configuration and 3.4 removed support for it after # which floating point timestamps are always supported. ver = sys.version_info[0:2] python_nanosecond_support = ver >= (3, 4) or (ver >= (2, 3) and os.stat_float_times()) # Minimal expected floating point difference used to account for # possible imprecise floating point number representations. We want # this number to be small (at least smaller than 0.0001) but still # large enough that we can be sure that increasing a floating point # value by 2 * eta guarantees the value read back will be increased by # at least eta. eta = 0.00005 stats_orig = os.stat(path) def test_time(diff): """Returns whether a timestamp difference is detectable.""" os.utime(path, (stats_orig.st_atime, stats_orig.st_mtime + diff)) return os.stat(path).st_mtime > stats_orig.st_mtime + eta # Test for nanosecond timestamp resolution support. if not minimum_resolution and python_nanosecond_support: if test_time(2 * eta): return 0 # Detect the filesystem timestamp resolution. Note that there is no # need to make this code 'as fast as possible' as, this function gets # called before having to sleep until the next detectable modification # timestamp value and that, since we already know nanosecond resolution # is not supported, will surely take longer than whatever we do here to # detect this minimal detectable modification timestamp resolution. step = 0.1 if not python_nanosecond_support: # If Python does not support nanosecond timestamp resolution we # know the minimum possible supported timestamp resolution is 1 # second. minimum_resolution = max(1, minimum_resolution) index = max(1, int(minimum_resolution / step)) while step * index < minimum_resolution: # Floating point number representation errors may cause our # initially calculated start index to be too small if calculated # directly. index += 1 while True: # Do not simply add up the steps to avoid cumulative floating point # number representation errors. next = step * index if next > 10: raise TestEnvironmentError("File systems with too coarse " "modification timestamp resolutions not supported.") if test_time(next): return next index += 1 def __read_file(self, name, exact=False): name = self.adjust_names(name)[0] result = "" try: if exact: result = self.read(name) else: result = self.read_and_strip(name).replace("\\", "/") except (IOError, IndexError): print "Note: could not open file", name self.fail_test(1) return result def __wait_for_time_change(self, path, touch, last_build_time): """ Wait until a newly assigned file system modification timestamp for the given path is large enough for the timestamp difference between it and the last build timestamp or the path's original file system modification timestamp (depending on the last_build_time flag) to be correctly recognized by both this Python based testing framework and the Boost Jam executable being tested. May optionally touch the given path to set its modification timestamp to the new value. """ assert self.last_build_timestamp or not last_build_time stats_orig = os.stat(path) if last_build_time: start_time = self.last_build_timestamp else: start_time = stats_orig.st_mtime build_resolution = self.__build_timestamp_resolution() assert build_resolution >= 0 # Check whether the current timestamp is already new enough. if stats_orig.st_mtime > start_time and (not build_resolution or stats_orig.st_mtime >= start_time + build_resolution): return resolution = self.__python_timestamp_resolution(path, build_resolution) assert resolution >= build_resolution # Implementation notes: # * Theoretically time.sleep() API might get interrupted too soon # (never actually encountered). # * We encountered cases where we sleep just long enough for the # filesystem's modifiction timestamp to change to the desired value, # but after waking up, the read timestamp is still just a tiny bit # too small (encountered on Windows). This is most likely caused by # imprecise floating point timestamp & sleep interval representation # used by Python. Note though that we never encountered a case where # more than one additional tiny sleep() call was needed to remedy # the situation. # * We try to wait long enough for the timestamp to change, but do not # want to waste processing time by waiting too long. The main # problem is that when we have a coarse resolution, the actual times # get rounded and we do not know the exact sleep time needed for the # difference between two such times to pass. E.g. if we have a 1 # second resolution and the original and the current file timestamps # are both 10 seconds then it could be that the current time is # 10.99 seconds and that we can wait for just one hundredth of a # second for the current file timestamp to reach its next value, and # using a longer sleep interval than that would just be wasting # time. while True: os.utime(path, None) c = os.stat(path).st_mtime if resolution: if c > start_time and (not build_resolution or c >= start_time + build_resolution): break if c <= start_time - resolution: # Move close to the desired timestamp in one sleep, but not # close enough for timestamp rounding to potentially cause # us to wait too long. if start_time - c > 5: if last_build_time: error_message = ("Last build time recorded as " "being a future event, causing a too long " "wait period. Something must have played " "around with the system clock.") else: error_message = ("Original path modification " "timestamp set to far into the future or " "something must have played around with the " "system clock, causing a too long wait " "period.\nPath: '%s'" % path) raise TestEnvironmentError(message) _sleep(start_time - c) else: # We are close to the desired timestamp so take baby sleeps # to avoid sleeping too long. _sleep(max(0.01, resolution / 10)) else: if c > start_time: break _sleep(max(0.01, start_time - c)) if not touch: os.utime(path, (stats_orig.st_atime, stats_orig.st_mtime)) class List: def __init__(self, s=""): elements = [] if s.__class__ is str: # Have to handle escaped spaces correctly. elements = s.replace("\ ", "\001").split() else: elements = s self.l = [e.replace("\001", " ") for e in elements] def __len__(self): return len(self.l) def __getitem__(self, key): return self.l[key] def __setitem__(self, key, value): self.l[key] = value def __delitem__(self, key): del self.l[key] def __str__(self): return str(self.l) def __repr__(self): return "%s.List(%r)" % (self.__module__, " ".join(self.l)) def __mul__(self, other): result = List() if not isinstance(other, List): other = List(other) for f in self: for s in other: result.l.append(f + s) return result def __rmul__(self, other): if not isinstance(other, List): other = List(other) return List.__mul__(other, self) def __add__(self, other): result = List() result.l = self.l[:] + other.l[:] return result def _contains_lines(data, lines): data_line_count = len(data) expected_line_count = reduce(lambda x, y: x + len(y), lines, 0) index = 0 for expected in lines: if expected_line_count > data_line_count - index: return False expected_line_count -= len(expected) index = _match_line_sequence(data, index, data_line_count - expected_line_count, expected) if index < 0: return False return True def _match_line_sequence(data, start, end, lines): if not lines: return start for index in xrange(start, end - len(lines) + 1): data_index = index for expected in lines: if not fnmatch.fnmatch(data[data_index], expected): break; data_index += 1 else: return data_index return -1 def _sleep(delay): if delay > 5: raise TestEnvironmentError("Test environment error: sleep period of " "more than 5 seconds requested. Most likely caused by a file with " "its modification timestamp set to sometime in the future.") time.sleep(delay) ############################################################################### # # Initialization. # ############################################################################### # Make os.stat() return file modification times as floats instead of integers # to get the best possible file timestamp resolution available. The exact # resolution depends on the underlying file system and the Python os.stat() # implementation. The better the resolution we achieve, the shorter we need to # wait for files we create to start getting new timestamps. # # Additional notes: # * os.stat_float_times() function first introduced in Python 2.3. and # suggested for deprecation in Python 3.3. # * On Python versions 2.5+ we do not need to do this as there os.stat() # returns floating point file modification times by default. # * Windows CPython implementations prior to version 2.5 do not support file # modification timestamp resolutions of less than 1 second no matter whether # these timestamps are returned as integer or floating point values. # * Python documentation states that this should be set in a program's # __main__ module to avoid affecting other libraries that might not be ready # to support floating point timestamps. Since we use no such external # libraries, we ignore this warning to make it easier to enable this feature # in both our single & multiple-test scripts. if (2, 3) <= sys.version_info < (2, 5) and not os.stat_float_times(): os.stat_float_times(True) # Quickie tests. Should use doctest instead. if __name__ == "__main__": assert str(List("foo bar") * "/baz") == "['foo/baz', 'bar/baz']" assert repr("foo/" * List("bar baz")) == "__main__.List('foo/bar foo/baz')" assert _contains_lines([], []) assert _contains_lines([], [[]]) assert _contains_lines([], [[], []]) assert _contains_lines([], [[], [], []]) assert not _contains_lines([], [[""]]) assert not _contains_lines([], [["a"]]) assert _contains_lines([""], []) assert _contains_lines(["a"], []) assert _contains_lines(["a", "b"], []) assert _contains_lines(["a", "b"], [[], [], []]) assert _contains_lines([""], [[""]]) assert not _contains_lines([""], [["a"]]) assert not _contains_lines(["a"], [[""]]) assert _contains_lines(["a", "", "b", ""], [["a"]]) assert _contains_lines(["a", "", "b", ""], [[""]]) assert _contains_lines(["a", "", "b"], [["b"]]) assert not _contains_lines(["a", "b"], [[""]]) assert not _contains_lines(["a", "", "b", ""], [["c"]]) assert _contains_lines(["a", "", "b", "x"], [["x"]]) data = ["1", "2", "3", "4", "5", "6", "7", "8", "9"] assert _contains_lines(data, [["1", "2"]]) assert not _contains_lines(data, [["2", "1"]]) assert not _contains_lines(data, [["1", "3"]]) assert not _contains_lines(data, [["1", "3"]]) assert _contains_lines(data, [["1"], ["2"]]) assert _contains_lines(data, [["1"], [], [], [], ["2"]]) assert _contains_lines(data, [["1"], ["3"]]) assert not _contains_lines(data, [["3"], ["1"]]) assert _contains_lines(data, [["3"], ["7"], ["8"]]) assert not _contains_lines(data, [["1"], ["3", "5"]]) assert not _contains_lines(data, [["1"], [""], ["5"]]) assert not _contains_lines(data, [["1"], ["5"], ["3"]]) assert not _contains_lines(data, [["1"], ["5", "3"]]) assert not _contains_lines(data, [[" 3"]]) assert not _contains_lines(data, [["3 "]]) assert not _contains_lines(data, [["3", ""]]) assert not _contains_lines(data, [["", "3"]]) print("tests passed")