Implement a --failed flag for the test command.

This will use ``tool_test_output.json`` to determine which tests failed the last time ``planemo test`` was executed and only run those previously failing tests. This should speed up test-driven development of tools.

Requires galaxyproject/galaxy#266.

Closes #145.

Author: jmchilton

planemo/commands/cmd_test.py

@@ -82,6 +82,15 @@
           "reports for more complete summary). Set to 'none' to disable "
           "completely.")
 )
+@click.option(
+    "--failed",
+    is_flag=True,
+    help="Re-run only failed tests. This command will read "
+         "tool_test_output.json to determine which tests failed so this "
+         "file must have been produced with the same set of tool ids "
+         "previously.",
+    default=False,
+)
 @options.galaxy_root_option()
 @options.install_galaxy_option()
 @options.no_cache_galaxy_option()
@@ -149,6 +158,7 @@ def cli(ctx, path, **kwds):
             html_report_file,
             xunit_report_file,
             structured_report_file,
+            failed=kwds["failed"],
         ).build()
         cmd = "; ".join([
             cd_to_galaxy_command,

planemo/galaxy_test.py

@@ -1,5 +1,6 @@
 """ Utilities for reasoning about Galaxy test results.
 """
+from __future__ import print_function
 from __future__ import absolute_import
 
 from collections import namedtuple
@@ -18,10 +19,12 @@ def __init__(
         html_report_file,
         xunit_report_file,
         structured_report_file,
+        failed=False,
     ):
         self.html_report_file = html_report_file
         self.xunit_report_file = xunit_report_file
         self.structured_report_file = structured_report_file
+        self.failed = failed
 
     def build(self):
         xunit_report_file = self.xunit_report_file
@@ -36,58 +39,45 @@ def build(self):
         else:
             sd_arg = ""
         tests = "functional.test_toolbox"
+        if self.failed:
+            sd = StructuredData(self.structured_report_file)
+            failed_ids = sd.failed_ids
+            tests = " ".join(failed_ids)
         return RUN_TESTS_CMD % (html_report_file, xunit_arg, sd_arg, tests)
 
 
-class GalaxyTestResults(object):
-    """ Class that combine the test-centric xunit output
-    with the Galaxy centric structured data output - and
-    abstracts away the difference (someday).
+class StructuredData(object):
+    """ Abstraction around Galaxy's structured test data output.
     """
 
-    def __init__(
-        self,
-        output_json_path,
-        output_xml_path,
-        output_html_path,
-        exit_code,
-    ):
-        self.output_html_path = output_html_path
-        self.exit_code = exit_code
+    def __init__(self, json_path):
+        self.json_path = json_path
         try:
-            output_json_f = open(output_json_path, "r")
-            structured_data = json.load(output_json_f)
-            structured_data_tests = structured_data["tests"]
+            with open(json_path, "r") as output_json_f:
+                structured_data = json.load(output_json_f)
+                structured_data_tests = structured_data["tests"]
         except Exception:
-            # Older Galaxy's will not support this option.
+            print("Warning: Targetting older Galaxy which did not "
+                  "produce a structured test results files.")
             structured_data = {}
             structured_data_tests = {}
         self.structured_data = structured_data
         self.structured_data_tests = structured_data_tests
-
         structured_data_by_id = {}
         for test in self.structured_data_tests:
             structured_data_by_id[test["id"]] = test["data"]
         self.structured_data_by_id = structured_data_by_id
+        self.has_details = "summary" in structured_data
+        if self.has_details:
+            self._read_summary()
 
-        if output_xml_path:
-            self.xunit_tree = parse_xunit_report(output_xml_path)
-            self.__merge_xunit()
-            self.has_details = True
-        else:
-            self.xunit_tree = ET.fromstring("<testsuite />")
-            self.has_details = False
-        try:
-            json.dump(self.structured_data, open(output_json_path, "w"))
-        except Exception:
-            pass
-
-    @property
-    def _xunit_root(self):
-        return self.xunit_tree.getroot()
+    def update(self):
+        with open(self.json_path, "w") as out_f:
+            json.dump(self.structured_data, out_f)
 
-    def __merge_xunit(self):
-        xunit_attrib = self._xunit_root.attrib
+    def merge_xunit(self, xunit_root):
+        self.has_details = True
+        xunit_attrib = xunit_root.attrib
         num_tests = int(xunit_attrib.get("tests", 0))
         num_failures = int(xunit_attrib.get("failures", 0))
         num_errors = int(xunit_attrib.get("errors", 0))
@@ -103,7 +93,7 @@ def __merge_xunit(self):
         self.num_tests = num_tests
         self.num_problems = num_skips + num_errors + num_failures
 
-        for testcase_el in self.xunit_testcase_elements:
+        for testcase_el in xunit_testcase_elements_from_root(xunit_root):
             test = case_id(testcase_el)
             test_data = self.structured_data_by_id.get(test.id)
             if not test_data:
@@ -121,14 +111,78 @@ def __merge_xunit(self):
                 status = "success"
             test_data["status"] = status
 
+    def _read_summary(self):
+        # TODO: read relevant data out of summary object.
+        pass
+
+    @property
+    def failed_ids(self):
+        ids = set([])
+        for test_data in self.structured_data_tests:
+            if test_data["data"]["status"] == "success":
+                continue
+            test_case = test_data["id"].replace(".test_toolbox.", ".test_toolbox:")
+            #test_case = test_case[0:test_case.find(".test_tool_")]
+            ids.add(test_case)
+        return ids
+
+
+class GalaxyTestResults(object):
+    """ Class that combine the test-centric xunit output
+    with the Galaxy centric structured data output - and
+    abstracts away the difference (someday).
+    """
+
+    def __init__(
+        self,
+        output_json_path,
+        output_xml_path,
+        output_html_path,
+        exit_code,
+    ):
+        self.output_html_path = output_html_path
+        self.exit_code = exit_code
+        sd = StructuredData(output_json_path)
+        self.sd = sd
+        self.structured_data = sd.structured_data
+        self.structured_data_tests = sd.structured_data_tests
+        self.structured_data_by_id = sd.structured_data_by_id
+
+        if output_xml_path:
+            self.xunit_tree = parse_xunit_report(output_xml_path)
+            sd.merge_xunit(self._xunit_root)
+        else:
+            self.xunit_tree = ET.fromstring("<testsuite />")
+        self.sd.update()
+
+    @property
+    def has_details(self):
+        return self.sd.has_details
+
+    @property
+    def num_tests(self):
+        return self.sd.num_tests
+
+    @property
+    def num_problems(self):
+        return self.sd.num_problems
+
+    @property
+    def _xunit_root(self):
+        return self.xunit_tree.getroot()
+
     @property
     def all_tests_passed(self):
-        return self.num_problems == 0
+        return self.sd.num_problems == 0
 
     @property
     def xunit_testcase_elements(self):
-        for testcase_el in find_cases(self._xunit_root):
-            yield testcase_el
+        return xunit_testcase_elements_from_root(self._xunit_root)
+
+
+def xunit_testcase_elements_from_root(xunit_root):
+    for testcase_el in find_cases(xunit_root):
+        yield testcase_el
 
 
 def parse_xunit_report(xunit_report_path):