pin the pytest pex interpreter path to the selected interpreter

src/python/pants/backend/python/tasks/pytest_prep.py

@@ -35,6 +35,7 @@ def __init__(self, interpreter, pex):
       pex_info.merge_pex_path(pex_path)  # We're now on the sys.path twice.
       PEXBuilder(pex_path, interpreter=interpreter, pex_info=pex_info).freeze()
       self._pex = PEX(pex=pex_path, interpreter=interpreter)
+      self._interpreter = interpreter
 
     @property
     def pex(self):
@@ -44,6 +45,14 @@ def pex(self):
       """
       return self._pex
 
+    @property
+    def interpreter(self):
+      """
+
+      :rtype: :class:`pex.interpreter.PythonInterpreter`
+      """
+      return self._interpreter
+
     @property
     def config_path(self):
       """Return the absolute path of the `pytest.ini` config file in this py.test binary.

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

@@ -70,6 +70,7 @@ def files_iter():
     return list(files_iter())
 
 
+# TODO: convert this into an enum!
 class PytestResult(TestResult):
   _SUCCESS_EXIT_CODES = (
     0,
@@ -482,6 +483,21 @@ def _test_runner(self, workdirs, test_targets, sources_map):
                                           pytest_binary.pex) as coverage_args:
         yield pytest_binary, [conftest] + coverage_args, get_pytest_rootdir
 
+  def _constrain_pytest_interpreter_search_path(self):
+    """Return an environment for invoking a pex which ensures the use of the selected interpreter.
+
+    When creating the merged pytest pex, we already have an interpreter, and we only invoke that pex
+    within a pants run, so we can be sure the selected interpreter will be available. Constraining
+    the interpreter search path at pex runtime ensures that any resolved requirements will be
+    compatible with the interpreter being used to invoke the merged pytest pex.
+    """
+    pytest_prep_binary_product = self.context.products.get_data(PytestPrep.PytestBinary)
+    chosen_interpreter_binary_path = pytest_prep_binary_product.interpreter.binary
+    return {
+      'PEX_PYTHON': chosen_interpreter_binary_path,
+      'PEX_PYTHON_PATH': chosen_interpreter_binary_path,
+    }
+
   def _do_run_tests_with_args(self, pex, args):
     try:
       env = dict(os.environ)
@@ -517,6 +533,8 @@ def _do_run_tests_with_args(self, pex, args):
       with self.context.new_workunit(name='run',
                                      cmd=' '.join(pex.cmdline(args)),
                                      labels=[WorkUnitLabel.TOOL, WorkUnitLabel.TEST]) as workunit:
+        # NB: Constrain the pex environment to ensure the use of the selected interpreter!
+        env.update(self._constrain_pytest_interpreter_search_path())
         rc = self.spawn_and_wait(pex, workunit=workunit, args=args, setsid=True, env=env)
         return PytestResult.rc(rc)
     except ErrorWhileTesting:
@@ -736,6 +754,8 @@ def _pex_run(self, pex, workunit_name, args, env):
     with self.context.new_workunit(name=workunit_name,
                                    cmd=' '.join(pex.cmdline(args)),
                                    labels=[WorkUnitLabel.TOOL, WorkUnitLabel.TEST]) as workunit:
+      # NB: Constrain the pex environment to ensure the use of the selected interpreter!
+      env.update(self._constrain_pytest_interpreter_search_path())
       process = self._spawn(pex, workunit, args, setsid=False, env=env)
       return process.wait()
 

src/python/pants/backend/python/tasks/python_execution_task_base.py

@@ -116,10 +116,14 @@ def create_pex(self, pex_info=None):
           # Add the extra requirements first, so they take precedence over any colliding version
           # in the target set's dependency closure.
           pexes = [extra_requirements_pex] + pexes
-        constraints = {constraint for rt in relevant_targets if is_python_target(rt)
-                       for constraint in PythonSetup.global_instance().compatibility_or_constraints(rt)}
 
-        with self.merged_pex(path, pex_info, interpreter, pexes, constraints) as builder:
+        unique_constraints = {
+          constraint for rt in relevant_targets if is_python_target(rt)
+          for constraint in PythonSetup.global_instance().compatibility_or_constraints(rt)
+        }
+        self.context.log.debug('unique_constraints:\n{}'.format(unique_constraints))
+
+        with self.merged_pex(path, pex_info, interpreter, pexes, unique_constraints) as builder:
           for extra_file in self.extra_files():
             extra_file.add_to(builder)
           builder.freeze()

tests/python/pants_test/backend/python/tasks/test_pytest_run.py

@@ -274,6 +274,18 @@ def test_use_two(self):
   dependencies = ["lib:core"],
   coverage = ["core"],
 )
+
+python_tests(
+  name = "py23-tests",
+  sources = ["py23_test_source.py"],
+  compatibility = ['CPython>=2.7,<4'],
+)
+
+python_tests(
+  name = "py3-and-more-tests",
+  sources = ["py3_and_more_test_source.py"],
+  compatibility = ['CPython>=3.6'],
+)
 '''
     )
 
@@ -359,6 +371,9 @@ def null():
         assert(False)
         """))
 
+    self.create_file('tests/py23_test_source.py', '')
+    self.create_file('tests/py3_and_more_test_source.py', '')
+
     self.create_file('tests/conftest.py', self._CONFTEST_CONTENT)
 
     self.app = self.target('tests:app')
@@ -375,6 +390,9 @@ def null():
     self.all = self.target('tests:all')
     self.all_with_cov = self.target('tests:all-with-coverage')
 
+    self.py23 = self.target('tests:py23-tests')
+    self.py3_and_more = self.target('tests:py3-and-more-tests')
+
   @ensure_cached(PytestRun, expected_num_artifacts=0)
   def test_error(self):
     """Test that a test that errors rather than fails shows up in ErrorWhileTesting."""
@@ -387,6 +405,10 @@ def test_error_outside_function(self):
     self.run_failing_tests(targets=[self.red, self.green, self.failure_outside_function],
                            failed_targets=[self.red, self.failure_outside_function])
 
+  @ensure_cached(PytestRun, expected_num_artifacts=1)
+  def test_succeeds_for_intersecting_unique_constraints(self):
+    self.run_tests(targets=[self.py23, self.py3_and_more])
+
   @ensure_cached(PytestRun, expected_num_artifacts=1)
   def test_green(self):
     self.run_tests(targets=[self.green])