Enhancements to Galaxy's functional testing framework as outlined in issue 490. https://bitbucket.org/galaxy/galaxy-central/issue/490/enabling-iterative-test-driven-development

lib/galaxy/tools/__init__.py

@@ -662,8 +662,31 @@ class Tool:
                     name = attrib.pop( 'name', None )
                     if name is None:
                         raise Exception( "Test output does not have a 'name'" )
+
+                    assert_elem = output_elem.find("assert_contents")
+                    assert_list = None
+                    # Trying to keep testing patch as localized as                                              
+                    # possible, this function should be relocated                                               
+                    # somewhere more conventional.
+                    def convert_elem(elem):
+                        """ Converts and XML element to a dictionary format, used by assertion checking code. """
+                        tag = elem.tag
+                        attributes = dict( elem.attrib )
+                        child_elems = list( elem.getchildren() )
+                        converted_children = []
+                        for child_elem in child_elems:
+                            converted_children.append( convert_elem(child_elem) )
+                        return {"tag" : tag, "attributes" : attributes, "children" : converted_children}
+                    
+                    if assert_elem is not None:
+                        assert_list = []
+                        for assert_child in list(assert_elem):
+                            assert_list.append(convert_elem(assert_child))
+
+
                     file = attrib.pop( 'file', None )
-                    if file is None:
+                    # File no longer required if an list of assertions was present.                             
+                    if assert_list is None and file is None:
                         raise Exception( "Test output does not have a 'file'")
                     attributes = {}
                     # Method of comparison
@@ -674,6 +697,9 @@ class Tool:
                     attributes['delta'] = int( attrib.pop( 'delta', '10000' ) ) 
                     attributes['sort'] = util.string_as_bool( attrib.pop( 'sort', False ) )
                     attributes['extra_files'] = []
+                    attributes['assert_list'] = assert_list
+
+
                     if 'ftype' in attrib:
                         attributes['ftype'] = attrib['ftype']
                     for extra in output_elem.findall( 'extra_files' ):

test/base/asserts/__init__.py

@@ -0,0 +1,76 @@
+import inspect
+import logging
+import sys
+log = logging.getLogger( __name__ )
+
+assertion_module_names = ['text', 'tabular', 'xml']
+
+# Code for loading modules containing assertion checking functions, to
+# create a new module of assertion functions, create the needed python
+# source file "test/base/asserts/<MODULE_NAME>.py" and add
+# <MODULE_NAME> to the list of assertion module names defined above. 
+assertion_modules = []
+for assertion_module_name in assertion_module_names:
+    full_assertion_module_name = 'base.asserts.' + assertion_module_name
+    log.debug(full_assertion_module_name)
+    try:
+        #Dynamically import module
+        __import__(full_assertion_module_name)
+        assertion_module = sys.modules[full_assertion_module_name]
+        assertion_modules.append(assertion_module)
+    except Exception, e:
+        log.exception( 'Failed to load assertion module: %s %s' % (assertion_module_name, str(e))) 
+
+def verify_assertions(data, assertion_description_list):
+    """ This function takes a list of assertions and a string to check
+    these assertions against. """
+    for assertion_description in assertion_description_list:
+        verify_assertion(data, assertion_description)
+
+def verify_assertion(data, assertion_description):
+    tag = assertion_description["tag"]
+    assert_function_name = "assert_" + tag
+    assert_function = None
+    for assertion_module in assertion_modules:
+        if hasattr(assertion_module, assert_function_name):
+            assert_function = getattr(assertion_module, assert_function_name)
+
+    if assert_function is None:
+        errmsg = "Unable to find test function associated with XML tag '%s'. Check your tool file syntax." % tag
+        raise AssertionError(errmsg)
+    
+    assert_function_args = inspect.getargspec(assert_function).args
+    args = {}
+    for attribute, value in assertion_description["attributes"].iteritems():
+        if attribute in assert_function_args:
+            args[attribute] = value
+
+    # Three special arguments automatically populated independently of
+    # tool XML attributes. output is passed in as the contents of the
+    # output file. verify_assertions_function is passed in as the
+    # verify_assertions function defined above, this allows
+    # recursively checking assertions on subsections of
+    # output. children is the parsed version of the child elements of
+    # the XML element describing this assertion. See
+    # assert_element_text in test/base/asserts/xml.py as an example of
+    # how to use verify_assertions_function and children in conjuction
+    # to apply assertion checking to a subset of the input. The parsed
+    # version of an elements child elements do not need to just define
+    # assertions, developers of assertion functions can also use the
+    # child elements in novel ways to define inputs the assertion
+    # checking function (for instance consider the following fictional
+    # assertion function for checking column titles of tabular output
+    # - <has_column_titles><with_name name="sequence"><with_name
+    # name="probability"></has_column_titles>.)
+    if "output" in assert_function_args:
+        args["output"] = data
+
+    if "verify_assertions_function" in assert_function_args:
+        args["verify_assertions_function"] = verify_assertions
+
+    if "children" in assert_function_args:
+        args["children"] = assertion_description["children"]
+        
+    # TODO: Verify all needed function arguments are specified.
+    assert_function(**args)
+    

test/base/asserts/tabular.py

@@ -0,0 +1,17 @@
+import re
+
+def get_first_line(output):
+    match = re.search("^(.*)$", output, flags = re.MULTILINE)
+    if match is None:
+        return None
+    else:
+        return match.group(1)
+
+def assert_has_n_columns(output, n, sep = '\t'):
+    """ Asserts the tabular output contains n columns. The optional
+    sep argument specifies the column seperator used to determine the
+    number of columns."""
+    n = int(n)
+    first_line = get_first_line(output)
+    assert first_line is not None, "Was expecting output with %d columns, but output was empty." % n
+    assert len(first_line.split(sep)) == n, "Output does not have %d columns." % n

test/base/asserts/text.py

@@ -0,0 +1,30 @@
+import re
+
+def assert_has_text(output, text):
+    """ Asserts specified output contains the substring specified by
+    the argument text."""
+    assert output.find(text) >= 0, "Output file did not contain expected text '%s' (ouptut '%s')" % (text, output)
+            
+def assert_not_has_text(output, text):
+    """ Asserts specified output does not contain the substring
+    specified the argument text."""
+    assert output.find(text) < 0, "Output file contains unexpected text '%s'" % text
+
+def assert_has_line(output, line):
+    """ Asserts the specified output contains the line specified the
+    argument line."""
+    match = re.search("^%s$" % re.escape(line), output, flags = re.MULTILINE)
+    assert match != None, "No line of output file was '%s' (output was '%s') " % (line, output)
+
+def assert_has_text_matching(output, expression):
+    """ Asserts the specified output contains text matching the
+    regular expression specified by the argument expression."""
+    match = re.search(expression, output)
+    assert match != None, "No text matching expression '%s' was found in output file." % expression
+
+def assert_has_line_matching(output, expression):
+    """ Asserts the specified output contains a line matching the
+    regular expression specified by the argument expression."""
+    match = re.search("^%s$" % expression, output, flags = re.MULTILINE)
+    assert match != None, "No line matching expression '%s' was found in output file." % expression
+

test/base/asserts/xml.py

@@ -0,0 +1,77 @@
+import elementtree.ElementTree
+import re
+
+# Helper functions used to work with XML output.
+def to_xml(output):
+  return elementtree.ElementTree.fromstring(output)
+
+def xml_find_text(output, path):
+  xml = to_xml(output)
+  text = xml.findtext(path)
+  return text
+
+def xml_find(output, path):
+  xml = to_xml(output)
+  return xml.find(path)
+
+def assert_is_valid_xml(output):
+  """ Simple assertion that just verifies the specified output
+  is valid XML."""
+  try:
+    to_xml(output)
+  except Exception, e:
+    # TODO: Narrow caught exception to just parsing failure   
+    raise AssertionError("Expected valid XML, but could not parse output. %s" % str(e))
+
+def assert_has_element_with_path(output, path):
+  """ Asserts the specified output has at least one XML element with a
+  path matching the specified path argument. Valid paths are the
+  simplified subsets of XPath implemented by elementtree (currently
+  Galaxy makes use of elementtree 1.2). See
+  http://effbot.org/zone/element-xpath.htm for more information."""
+  if xml_find(output, path) is None:
+    errmsg = "Expected to find XML element matching expression %s, not such match was found." % path
+    raise AssertionError(errmsg)
+
+def assert_has_n_elements_with_path(output, path, n):
+  """ Asserts the specified output has exactly n elements matching the
+  path specified."""
+  xml = to_xml(output)
+  n = int(n)
+  num_elements = len(xml.findall(path))
+  if num_elements != n:
+    errmsg = "Expected to find %d elements with path %s, but %d were found." % (n, path, num_elements)
+    raise AssertionError(errmsg)
+
+def assert_element_text_matches(output, path, expression):
+  """ Asserts the text of the first element matching the specified
+  path matches the specified regular expression."""
+  text = xml_find_text(output, path)
+  if re.match(expression, text) is None:
+    errmsg = "Expected element with path '%s' to contain text matching '%s', instead text '%s' was found." % (path, text, actual_text)
+    raise AssertionError(errmsg)
+
+def assert_element_text_is(output, path, text):
+  """ Asserts the text of the first element matching the specified
+  path matches exactly the specified text. """
+  assert_element_text_matches(output, path, re.escape(text))
+
+def assert_attribute_matches(output, path, attribute, expression):
+  """ Asserts the specified attribute of the first element matching
+  the specified path matches the specified regular expression."""
+  xml = xml_find(output, path)
+  attribute_value = xml.attrib[attribute]
+  if re.match(expression, attribute_value) is None:
+    errmsg = "Expected attribute '%s' on element with path '%s' to match '%s', instead attribute value was '%s'." % (attribute, path, expression, attribute_value)
+    raise AssertionError(errmsg)
+  
+def assert_attribute_is(output, path, attribute, text):
+  """ Asserts the specified attribute of the first element matching
+  the specified path matches exactly the specified text."""
+  assert_attribute_matches(output, path, attribute, re.escape(text))
+  
+def assert_element_text(output, path, verify_assertions_function, children):
+  """ Recursively checks the specified assertions against the text of
+  the first element matching the specified path."""
+  text = xml_find_text(output, path)
+  verify_assertions_function(text, children)

test/base/twilltestcase.py

@@ -11,6 +11,7 @@ pkg_resources.require( "elementtree" )
 from elementtree import ElementTree
 from galaxy.web import security
 from galaxy.web.framework.helpers import iff
+from base.asserts import verify_assertions
 
 buffer = StringIO.StringIO()
 
@@ -635,7 +636,7 @@ class TwillTestCase( unittest.TestCase ):
             elem = elems[0]
         self.assertTrue( hid )
         self._assert_dataset_state( elem, 'ok' )
-        if self.is_zipped( filename ):
+        if filename is not None and self.is_zipped( filename ):
             errmsg = 'History item %s is a zip archive which includes invalid files:\n' % hid
             zip_file = zipfile.ZipFile( filename, "r" )
             name = zip_file.namelist()[0]
@@ -647,50 +648,59 @@ class TwillTestCase( unittest.TestCase ):
                 if ext != test_ext:
                     raise AssertionError( errmsg )
         else:
-            local_name = self.get_filename( filename )
-            temp_name = self.makeTfname(fname = filename)
             self.home()
             self.visit_page( "display?hid=" + hid )
             data = self.last_page()
-            file( temp_name, 'wb' ).write(data)
-            if self.keepOutdir > '':
-                ofn = os.path.join(self.keepOutdir,os.path.basename(local_name))
-                shutil.copy(temp_name,ofn)
-                log.debug('## GALAXY_TEST_SAVE=%s. saved %s' % (self.keepOutdir,ofn))
-            try:
-                # have to nest try-except in try-finally to handle 2.4
+            assert_list = attributes["assert_list"]
+            if assert_list is not None:
                 try:
-                    if attributes is None:
-                        attributes = {}
-                    compare = attributes.get( 'compare', 'diff' )
-                    if attributes.get( 'ftype', None ) == 'bam':
-                        local_fh, temp_name = self._bam_to_sam( local_name, temp_name )
-                        local_name = local_fh.name
-                    extra_files = attributes.get( 'extra_files', None )
-                    if compare == 'diff':
-                        self.files_diff( local_name, temp_name, attributes=attributes )
-                    elif compare == 're_match':
-                        self.files_re_match( local_name, temp_name, attributes=attributes )
-                    elif compare == 're_match_multiline':
-                        self.files_re_match_multiline( local_name, temp_name, attributes=attributes )
-    	            elif compare == 'sim_size':
-    	                delta = attributes.get('delta','100')
-    	                s1 = len(data)
-    	                s2 = os.path.getsize(local_name)
-    	                if abs(s1-s2) > int(delta):
-    	                   raise Exception, 'Files %s=%db but %s=%db - compare (delta=%s) failed' % (temp_name,s1,local_name,s2,delta)
-                    elif compare == "contains":
-                        self.files_contains( local_name, temp_name, attributes=attributes )
-                    else:
-                        raise Exception, 'Unimplemented Compare type: %s' % compare
-                    if extra_files:
-                        self.verify_extra_files_content( extra_files, elem.get( 'id' ) )
+                    verify_assertions(data, assert_list)
                 except AssertionError, err:
-                    errmsg = 'History item %s different than expected, difference (using %s):\n' % ( hid, compare )
+                    errmsg = 'History item %s different than expected\n' % (hid)
                     errmsg += str( err )
                     raise AssertionError( errmsg )
-            finally:
-                os.remove( temp_name )
+            if filename is not None:
+               local_name = self.get_filename( filename )
+               temp_name = self.makeTfname(fname = filename)
+               file( temp_name, 'wb' ).write(data)
+               if self.keepOutdir > '':
+                   ofn = os.path.join(self.keepOutdir,os.path.basename(local_name))
+                   shutil.copy(temp_name,ofn)
+                   log.debug('## GALAXY_TEST_SAVE=%s. saved %s' % (self.keepOutdir,ofn))
+               try:
+                   # have to nest try-except in try-finally to handle 2.4
+                   try:
+                       if attributes is None:
+                           attributes = {}
+                       compare = attributes.get( 'compare', 'diff' )
+                       if attributes.get( 'ftype', None ) == 'bam':
+                           local_fh, temp_name = self._bam_to_sam( local_name, temp_name )
+                           local_name = local_fh.name
+                       extra_files = attributes.get( 'extra_files', None )
+                       if compare == 'diff':
+                           self.files_diff( local_name, temp_name, attributes=attributes )
+                       elif compare == 're_match':
+                           self.files_re_match( local_name, temp_name, attributes=attributes )
+                       elif compare == 're_match_multiline':
+                           self.files_re_match_multiline( local_name, temp_name, attributes=attributes )
+                       elif compare == 'sim_size':
+                           delta = attributes.get('delta','100')
+                           s1 = len(data)
+                           s2 = os.path.getsize(local_name)
+                           if abs(s1-s2) > int(delta):
+                               raise Exception, 'Files %s=%db but %s=%db - compare (delta=%s) failed' % (temp_name,s1,local_name,s2,delta)
+                       elif compare == "contains":
+                           self.files_contains( local_name, temp_name, attributes=attributes )
+                       else:
+                           raise Exception, 'Unimplemented Compare type: %s' % compare
+                       if extra_files:
+                           self.verify_extra_files_content( extra_files, elem.get( 'id' ) )
+                   except AssertionError, err:
+                       errmsg = 'History item %s different than expected, difference (using %s):\n' % ( hid, compare )
+                       errmsg += str( err )
+                       raise AssertionError( errmsg )
+               finally:
+                   os.remove( temp_name )
 
     def _bam_to_sam( self, local_name, temp_name ):
         temp_local = tempfile.NamedTemporaryFile( suffix='.sam', prefix='local_bam_converted_to_sam_' )