Fix test invalidation.

Previously failures escaped the `invalidated` context as return values,
subverting the invalidation mechanism. Fix this and enhance the `Task`
testing infrastructure to properly engage `Task` fingerprinting which
the `PytestRun` test now relies upon to pass.

src/python/pants/backend/python/tasks2/pytest_run.py

@@ -57,6 +57,13 @@ def rc(cls, value):
     exit_code = cls._map_exit_code(value)
     return PytestResult('SUCCESS' if exit_code == 0 else 'FAILURE', rc=exit_code)
 
+  @classmethod
+  def from_error(cls, error):
+    if not isinstance(error, TaskError):
+      raise AssertionError('Can only synthesize a {} from a TaskError, given a {}'
+                           .format(cls.__name__, type(error).__name__))
+    return cls.rc(error.exit_code).with_failed_targets(error.failed_targets)
+
   def with_failed_targets(self, failed_targets):
     return PytestResult(self._msg, self._rc, failed_targets)
 
@@ -76,6 +83,17 @@ def success(self):
   def failed_targets(self):
     return self._failed_targets
 
+  def checked(self):
+    """Raise if this result was unsuccessful and otherwise return this result unchanged.
+
+    :returns: this instance if successful
+    :rtype: :class:`PytestResult`
+    :raises: :class:`ErrorWhileTesting` if this result represents a failure
+    """
+    if not self.success:
+      raise ErrorWhileTesting(exit_code=self._rc, failed_targets=self._failed_targets)
+    return self
+
 
 class PytestRun(TestRunnerTaskMixin, Task):
 
@@ -498,7 +516,10 @@ def _run_tests(self, targets):
 
     results = {}
     for partition in partitions:
-      rv = self._do_run_tests(partition)
+      try:
+        rv = self._do_run_tests(partition)
+      except ErrorWhileTesting as e:
+        rv = PytestResult.from_error(e)
       results[partition] = rv
       if not rv.success and self.get_options().fail_fast:
         break
@@ -520,7 +541,8 @@ def _do_run_tests(self, targets):
                           invalidate_dependents=True) as invalidation_check:
 
       invalid_tgts = [tgt for vts in invalidation_check.invalid_vts for tgt in vts.targets]
-      return self._run_pytest(invalid_tgts)
+      result = self._run_pytest(invalid_tgts)
+      return result.checked()
 
   def _run_pytest(self, targets):
     if not targets:

src/python/pants/option/parser.py

@@ -148,7 +148,7 @@ def parse_args(self, flags, namespace):
     mutex_map = defaultdict(list)
     for args, kwargs in self._unnormalized_option_registrations_iter():
       self._validate(args, kwargs)
-      dest = kwargs.get('dest') or self._select_dest(args)
+      dest = self.parse_dest(*args, **kwargs)
 
       # Compute the values provided on the command line for this option.  Note that there may be
       # multiple values, for any combination of the following reasons:
@@ -242,7 +242,7 @@ def option_registrations_iter(self):
     """
     def normalize_kwargs(args, orig_kwargs):
       nkwargs = copy.copy(orig_kwargs)
-      dest = nkwargs.get('dest') or self._select_dest(args)
+      dest = self.parse_dest(*args, **nkwargs)
       nkwargs['dest'] = dest
       if not ('default' in nkwargs and isinstance(nkwargs['default'], RankedValue)):
         nkwargs['default'] = self._compute_value(dest, nkwargs, [])
@@ -395,11 +395,17 @@ def _existing_scope(self, arg):
 
   _ENV_SANITIZER_RE = re.compile(r'[.-]')
 
-  def _select_dest(self, args):
-    """Select the dest name for the option.
+  @staticmethod
+  def parse_dest(*args, **kwargs):
+    """Select the dest name for an option registration.
 
-    '--foo-bar' -> 'foo_bar' and '-x' -> 'x'.
+    If an explicit `dest` is specified, returns that and otherwise derives a default from the
+    option flags where '--foo-bar' -> 'foo_bar' and '-x' -> 'x'.
     """
+    explicit_dest = kwargs.get('dest')
+    if explicit_dest:
+      return explicit_dest
+
     arg = next((a for a in args if a.startswith('--')), args[0])
     return arg.lstrip('-').replace('-', '_')
 

tests/python/pants_test/option/util/fakes.py

@@ -9,7 +9,9 @@
 
 from pants.option.arg_splitter import GLOBAL_SCOPE
 from pants.option.global_options import GlobalOptionsRegistrar
+from pants.option.option_util import is_list_option
 from pants.option.optionable import Optionable
+from pants.option.parser import Parser
 from pants.option.parser_hierarchy import enclosing_scope
 from pants.option.ranked_value import RankedValue
 
@@ -96,6 +98,8 @@ def create_options(options, passthru_args=None):
   :param dict options: A dict of scope -> (dict of option name -> value).
   :returns: An fake `Options` object encapsulating the given scoped options.
   """
+  fingerprintable = defaultdict(dict)
+
   class FakeOptions(object):
     def for_scope(self, scope):
       scoped_options = options[scope]
@@ -120,8 +124,20 @@ def items(self):
     def scope_to_flags(self):
       return {}
 
+    def register(self, scope, *args, **kwargs):
+      fingerprint = kwargs.get('fingerprint', False)
+      if fingerprint:
+        option_name = Parser.parse_dest(*args, **kwargs)
+        if is_list_option(kwargs):
+          val_type = kwargs.get('member_type', str)
+        else:
+          val_type = kwargs.get('type', str)
+        fingerprintable[scope][option_name] = val_type
+
     def get_fingerprintable_for_scope(self, scope):
-      return []
+      option_values = self.for_scope(scope)
+      return [(option_type, option_values[option_name])
+              for option_name, option_type in fingerprintable[scope].items()]
 
     def __getitem__(self, key):
       return self.for_scope(key)

tests/python/pants_test/tasks/task_test_base.py

@@ -5,6 +5,7 @@
 from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
                         unicode_literals, with_statement)
 
+import functools
 import os
 import subprocess
 from contextlib import closing
@@ -144,6 +145,8 @@ def create_task(self, context, workdir=None):
     """
     :API: public
     """
+    register_func = functools.partial(context.options.register, self.options_scope)
+    self._testing_task_type.register_options(register_func)
     return self._testing_task_type(context, workdir or self._test_workdir)