diff --git a/src/python/pants/backend/python/tasks/gather_sources.py b/src/python/pants/backend/python/tasks/gather_sources.py
index 63548b283ce..2baa57eaaef 100644
--- a/src/python/pants/backend/python/tasks/gather_sources.py
+++ b/src/python/pants/backend/python/tasks/gather_sources.py
@@ -6,13 +6,14 @@
                         unicode_literals, with_statement)
 
 import os
+from collections import OrderedDict
 
 from pex.interpreter import PythonInterpreter
 from pex.pex import PEX
 from pex.pex_builder import PEXBuilder
 
 from pants.backend.python.tasks.pex_build_util import (dump_sources, has_python_sources,
-                                                       has_resources)
+                                                       has_resources, is_python_target)
 from pants.invalidation.cache_manager import VersionedTargetSet
 from pants.task.task import Task
 from pants.util.dirutil import safe_concurrent_creation
@@ -21,19 +22,44 @@
 class GatherSources(Task):
   """Gather local Python sources.
 
-  Creates an (unzipped) PEX on disk containing the local Python sources.
-  This PEX can be merged with a requirements PEX to create a unified Python environment
+  Creates one or more (unzipped) PEXs on disk containing the local Python sources.
+  These PEXes can be merged with a requirements PEX to create a unified Python environment
   for running the relevant python code.
   """
-  PYTHON_SOURCES = 'python_sources'
+
+  class PythonSources(object):
+    """A mapping of unzipped source PEXs by the targets whose sources the PEXs contain."""
+
+    class UnmappedTargetError(Exception):
+      """Indicates that no python source pex could be found for a given target."""
+
+    def __init__(self, pex_by_target_base):
+      self._pex_by_target_base = pex_by_target_base
+
+    def for_target(self, target):
+      """Return the unzipped PEX containing the given target's sources.
+
+      :returns: An unzipped PEX containing at least the given target's sources.
+      :rtype: :class:`pex.pex.PEX`
+      :raises: :class:`GatherSources.PythonSources.UnmappedTargetError` if no pex containing the
+               given target's sources could be found.
+      """
+      pex = self._pex_by_target_base.get(target.target_base)
+      if pex is None:
+        raise self.UnmappedTargetError()
+      return pex
+
+    def all(self):
+      """Return all the unzipped source PEXs needed for this round."""
+      return self._pex_by_target_base.values()
 
   @classmethod
   def implementation_version(cls):
-    return super(GatherSources, cls).implementation_version() + [('GatherSources', 3)]
+    return super(GatherSources, cls).implementation_version() + [('GatherSources', 4)]
 
   @classmethod
   def product_types(cls):
-    return [cls.PYTHON_SOURCES]
+    return [cls.PythonSources]
 
   @classmethod
   def prepare(cls, options, round_manager):
@@ -41,12 +67,55 @@ def prepare(cls, options, round_manager):
     round_manager.require_data('python')  # For codegen.
 
   def execute(self):
-    targets = self.context.targets(predicate=lambda t: has_python_sources(t) or has_resources(t))
     interpreter = self.context.products.get_data(PythonInterpreter)
 
-    with self.invalidated(targets) as invalidation_check:
-      pex = self._get_pex_for_versioned_targets(interpreter, invalidation_check.all_vts)
-      self.context.products.register_data(self.PYTHON_SOURCES, pex)
+    pex_by_target_base = OrderedDict()  # Preserve ~PYTHONPATH ordering over pexes.
+    for target_base, targets in self._iter_targets_by_base():
+      with self.invalidated(targets) as invalidation_check:
+        pex = self._get_pex_for_versioned_targets(interpreter, invalidation_check.all_vts)
+        pex_by_target_base[target_base] = pex
+    self.context.products.register_data(self.PythonSources, self.PythonSources(pex_by_target_base))
+
+  def _iter_targets_by_base(self):
+    # N.B: Files and Resources targets belong with the consuming (dependee) targets so that those
+    # targets can be ensured of access to the files in their PEX chroot. This means a given Files
+    # or Resources target could be embedded in multiple pexes.
+
+    context = self.context
+    python_target_addresses = [p.address for p in context.targets(predicate=is_python_target)]
+
+    targets_by_base = OrderedDict()  # Preserve ~PYTHONPATH ordering over source roots.
+    resource_targets = set()
+
+    def collect_source_targets(target):
+      if has_python_sources(target):
+        targets = targets_by_base.get(target.target_base)
+        if targets is None:
+          targets = set()
+          targets_by_base[target.target_base] = targets
+        targets.add(target)
+      elif has_resources(target):
+        resource_targets.add(target)
+
+    build_graph = context.build_graph
+    build_graph.walk_transitive_dependency_graph(addresses=python_target_addresses,
+                                                 work=collect_source_targets)
+
+    for resource_target in resource_targets:
+      dependees = build_graph.transitive_dependees_of_addresses([resource_target.address])
+      for target_base, targets in targets_by_base.items():
+        for dependee in dependees:
+          if dependee in targets:
+            # N.B.: This can add the resource to too many pexes. A canonical example is
+            # test -> lib -> resource where test and lib have separate source roots. In this case
+            # the resource is added to both the test pex and the lib pex and it's only needed in the
+            # lib pex. The upshot is we allow python code access to undeclared (ie: indirect)
+            # resource dependencies which is no worse than historical precedent, but could be
+            # improved with a more complex algorithm.
+            targets.add(resource_target)
+            break
+
+    return targets_by_base.items()
 
   def _get_pex_for_versioned_targets(self, interpreter, versioned_targets):
     if versioned_targets:
diff --git a/src/python/pants/backend/python/tasks/pex_build_util.py b/src/python/pants/backend/python/tasks/pex_build_util.py
index d05b1a954f7..bb3b0f2e514 100644
--- a/src/python/pants/backend/python/tasks/pex_build_util.py
+++ b/src/python/pants/backend/python/tasks/pex_build_util.py
@@ -23,11 +23,15 @@
 from pants.python.python_repos import PythonRepos
 
 
-def has_python_sources(tgt):
+def is_python_target(tgt):
   # We'd like to take all PythonTarget subclasses, but currently PythonThriftLibrary and
   # PythonAntlrLibrary extend PythonTarget, and until we fix that (which we can't do until
   # we remove the old python pipeline entirely) we want to ignore those target types here.
-  return isinstance(tgt, (PythonLibrary, PythonTests, PythonBinary)) and tgt.has_sources()
+  return isinstance(tgt, (PythonLibrary, PythonTests, PythonBinary))
+
+
+def has_python_sources(tgt):
+  return is_python_target(tgt) and tgt.has_sources()
 
 
 def has_resources(tgt):
diff --git a/src/python/pants/backend/python/tasks/pytest_run.py b/src/python/pants/backend/python/tasks/pytest_run.py
index 3a6a8c8bbc1..dc41fab548f 100644
--- a/src/python/pants/backend/python/tasks/pytest_run.py
+++ b/src/python/pants/backend/python/tasks/pytest_run.py
@@ -11,11 +11,13 @@
 import time
 import traceback
 import uuid
+from collections import OrderedDict
 from contextlib import contextmanager
 from textwrap import dedent
 
 from six import StringIO
 from six.moves import configparser
+from twitter.common.collections import OrderedSet
 
 from pants.backend.python.targets.python_tests import PythonTests
 from pants.backend.python.tasks.gather_sources import GatherSources
@@ -192,12 +194,6 @@ def _debug(self):
     return self.get_options().level == 'debug'
 
   def _generate_coverage_config(self, source_mappings):
-    # For the benefit of macos testing, add the 'real' path the directory as an equivalent.
-    def add_realpath(path):
-      realpath = os.path.realpath(path)
-      if realpath != canonical and realpath not in alternates:
-        realpaths.add(realpath)
-
     cp = configparser.SafeConfigParser()
     cp.readfp(StringIO(self.DEFAULT_COVERAGE_CONFIG))
 
@@ -205,15 +201,16 @@ def add_realpath(path):
     # coverage data files into canonical form.
     # See the "[paths]" entry here: http://nedbatchelder.com/code/coverage/config.html for details.
     cp.add_section('paths')
-    for canonical, alternates in source_mappings.items():
+    for canonical, alternate in source_mappings.items():
       key = canonical.replace(os.sep, '.')
-      realpaths = set()
-      add_realpath(canonical)
-      for path in alternates:
-        add_realpath(path)
-      cp.set('paths',
-             key,
-             self._format_string_list([canonical] + list(alternates) + list(realpaths)))
+
+      # For the benefit of macos testing, add the 'real' paths as equivalents.
+      paths = OrderedSet([canonical,
+                          alternate,
+                          os.path.realpath(canonical),
+                          os.path.realpath(alternate)])
+
+      cp.set('paths', key, self._format_string_list(paths))
 
     # See the debug options here: http://nedbatchelder.com/code/coverage/cmd.html#cmd-run-debug
     if self._debug:
@@ -250,14 +247,15 @@ def _maybe_emit_coverage_data(self, workdirs, targets, pex):
       yield []
       return
 
-    pex_src_root = os.path.relpath(self._source_chroot_path, get_buildroot())
+    def pex_src_root(tgt):
+      return os.path.relpath(self._source_chroot_path([tgt]), get_buildroot())
 
     source_mappings = {}
     for target in targets:
       libs = (tgt for tgt in target.closure()
               if tgt.has_sources('.py') and not isinstance(tgt, PythonTests))
       for lib in libs:
-        source_mappings[lib.target_base] = [pex_src_root]
+        source_mappings[lib.target_base] = pex_src_root(lib)
 
     def ensure_trailing_sep(path):
       return path if path.endswith(os.path.sep) else path + os.path.sep
@@ -287,13 +285,13 @@ def compute_coverage_sources(tgt):
           rel_source = os.path.relpath(source, get_buildroot())
           rel_source = ensure_trailing_sep(rel_source)
           found_target_base = False
-          for target_base in source_mappings:
+          for target_base, pex_root in source_mappings.items():
             prefix = ensure_trailing_sep(target_base)
             if rel_source.startswith(prefix):
               # ... rel_source will match on prefix=src/python/ ...
               suffix = rel_source[len(prefix):]
               # ... suffix will equal foo/bar ...
-              coverage_sources.append(os.path.join(pex_src_root, suffix))
+              coverage_sources.append(os.path.join(pex_root, suffix))
               found_target_base = True
               # ... and we end up appending <pex_src_root>/foo/bar to the coverage_sources.
               break
@@ -312,8 +310,11 @@ def compute_coverage_sources(tgt):
         env = {
           'PEX_MODULE': 'coverage.cmdline:main'
         }
-        def pex_run(arguments):
-          return self._pex_run(pex, workunit_name='coverage', args=arguments, env=env)
+        def coverage_run(subcommand, arguments):
+          return self._pex_run(pex,
+                               workunit_name='coverage-{}'.format(subcommand),
+                               args=[subcommand] + arguments,
+                               env=env)
 
         # On failures or timeouts, the .coverage file won't be written.
         if not os.path.exists('.coverage'):
@@ -323,13 +324,15 @@ def pex_run(arguments):
           # This swaps the /tmp pex chroot source paths for the local original source paths
           # the pex was generated from and which the user understands.
           shutil.move('.coverage', '.coverage.raw')
-          pex_run(['combine', '--rcfile', coverage_rc])
-          pex_run(['report', '-i', '--rcfile', coverage_rc])
+          # N.B.: This transforms the contents of .coverage.raw and moves it back into .coverage.
+          coverage_run('combine', ['--rcfile', coverage_rc])
+
+          coverage_run('report', ['-i', '--rcfile', coverage_rc])
 
           coverage_workdir = workdirs.coverage_path
-          pex_run(['html', '-i', '--rcfile', coverage_rc, '-d', coverage_workdir])
+          coverage_run('html', ['-i', '--rcfile', coverage_rc, '-d', coverage_workdir])
           coverage_xml = os.path.join(coverage_workdir, 'coverage.xml')
-          pex_run(['xml', '-i', '--rcfile', coverage_rc, '-o', coverage_xml])
+          coverage_run('xml', ['-i', '--rcfile', coverage_rc, '-o', coverage_xml])
 
   def _get_shard_conftest_content(self):
     shard_spec = self.get_options().test_shard
@@ -386,7 +389,7 @@ def _get_conftest_content(self, sources_map):
       import pytest
 
       # Map from source path relative to chroot -> source path relative to buildroot.
-      _SOURCES_MAP = {}
+      _SOURCES_MAP = {!r}
 
       @pytest.hookimpl(hookwrapper=True)
       def pytest_runtest_protocol(item, nextitem):
@@ -400,7 +403,7 @@ def pytest_runtest_protocol(item, nextitem):
           yield
         finally:
           item._nodeid = real_nodeid
-    """.format(sources_map))
+    """.format(dict(sources_map)))
     # Add in the sharding conftest, if any.
     shard_conftest_content = self._get_shard_conftest_content()
     return (console_output_conftest_content + shard_conftest_content).encode('utf8')
@@ -463,7 +466,7 @@ def _do_run_tests_with_args(self, pex, args):
       return PytestResult.exception()
 
   def _map_relsrc_to_targets(self, targets):
-    pex_src_root = os.path.relpath(self._source_chroot_path, get_buildroot())
+    pex_src_root = os.path.relpath(self._source_chroot_path(targets), get_buildroot())
     # First map chrooted sources back to their targets.
     relsrc_to_target = {os.path.join(pex_src_root, src): target for target in targets
       for src in target.sources_relative_to_source_root()}
@@ -505,7 +508,15 @@ def _iter_partitions(self, targets):
     # TODO(John Sirois): Consume `py.test` pexes matched to the partitioning in effect after
     # https://github.com/pantsbuild/pants/pull/4638 lands.
     if self.get_options().fast:
-      yield tuple(targets)
+      targets_by_target_base = OrderedDict()
+      for target in targets:
+        targets_for_base = targets_by_target_base.get(target.target_base)
+        if targets_for_base is None:
+          targets_for_base = []
+          targets_by_target_base[target.target_base] = targets_for_base
+        targets_for_base.append(target)
+      for targets in targets_by_target_base.values():
+        yield tuple(targets)
     else:
       for target in targets:
         yield (target,)
@@ -685,10 +696,10 @@ def _run_pytest(self, workdirs, targets):
     if self._run_in_chroot:
       path_func = lambda rel_src: rel_src
     else:
-      source_chroot = os.path.relpath(self._source_chroot_path, get_buildroot())
+      source_chroot = os.path.relpath(self._source_chroot_path(targets), get_buildroot())
       path_func = lambda rel_src: os.path.join(source_chroot, rel_src)
 
-    sources_map = {}  # Path from chroot -> Path from buildroot.
+    sources_map = OrderedDict()  # Path from chroot -> Path from buildroot.
     for t in targets:
       for p in t.sources_relative_to_source_root():
         sources_map[path_func(p)] = os.path.join(t.target_base, p)
@@ -726,7 +737,8 @@ def _run_pytest(self, workdirs, targets):
       if os.path.exists(junitxml_path):
         os.unlink(junitxml_path)
 
-      result = self._do_run_tests_with_args(pex, args)
+      with self._maybe_run_in_chroot(targets):
+        result = self._do_run_tests_with_args(pex, args)
 
       # There was a problem prior to test execution preventing junit xml file creation so just let
       # the failure result bubble.
@@ -748,9 +760,16 @@ def parse_error_handler(parse_error):
 
       return result.with_failed_targets(failed_targets)
 
-  @memoized_property
-  def _source_chroot_path(self):
-    return self.context.products.get_data(GatherSources.PYTHON_SOURCES).path()
+  def _source_chroot_path(self, targets):
+    if len(targets) > 1:
+      target_bases = {target.target_base for target in targets}
+      assert len(target_bases) == 1, ('Expected targets to live in the same source root, given '
+                                      'targets living under the following source roots: {}'
+                                      .format(', '.join(sorted(target_bases))))
+    representative_target = targets[0]
+
+    python_sources = self.context.products.get_data(GatherSources.PythonSources)
+    return python_sources.for_target(representative_target).path()
 
   def _pex_run(self, pex, workunit_name, args, env):
     with self.context.new_workunit(name=workunit_name,
@@ -764,21 +783,20 @@ def _run_in_chroot(self):
     return self.get_options().chroot
 
   @contextmanager
-  def _maybe_run_in_chroot(self):
+  def _maybe_run_in_chroot(self, targets):
     if self._run_in_chroot:
-      with pushd(self._source_chroot_path):
+      with pushd(self._source_chroot_path(targets)):
         yield
     else:
       yield
 
   def _spawn(self, pex, workunit, args, setsid=False, env=None):
-    with self._maybe_run_in_chroot():
-      env = env or {}
-      process = pex.run(args,
-                        with_chroot=False,  # We handle chrooting ourselves.
-                        blocking=False,
-                        setsid=setsid,
-                        env=env,
-                        stdout=workunit.output('stdout'),
-                        stderr=workunit.output('stderr'))
-      return SubprocessProcessHandler(process)
+    env = env or {}
+    process = pex.run(args,
+                      with_chroot=False,  # We handle chrooting ourselves.
+                      blocking=False,
+                      setsid=setsid,
+                      env=env,
+                      stdout=workunit.output('stdout'),
+                      stderr=workunit.output('stderr'))
+    return SubprocessProcessHandler(process)
diff --git a/src/python/pants/backend/python/tasks/python_execution_task_base.py b/src/python/pants/backend/python/tasks/python_execution_task_base.py
index a4e76755280..feff3b475cc 100644
--- a/src/python/pants/backend/python/tasks/python_execution_task_base.py
+++ b/src/python/pants/backend/python/tasks/python_execution_task_base.py
@@ -90,7 +90,7 @@ def prepare(cls, options, round_manager):
     super(PythonExecutionTaskBase, cls).prepare(options, round_manager)
     round_manager.require_data(PythonInterpreter)
     round_manager.require_data(ResolveRequirements.REQUIREMENTS_PEX)
-    round_manager.require_data(GatherSources.PYTHON_SOURCES)
+    round_manager.require_data(GatherSources.PythonSources)
 
   def extra_requirements(self):
     """Override to provide extra requirements needed for execution.
@@ -121,10 +121,9 @@ def create_pex(self, pex_info=None):
       # Note that we check for the existence of the directory, instead of for invalid_vts,
       # to cover the empty case.
       if not os.path.isdir(path):
-        pexes = [
-          self.context.products.get_data(ResolveRequirements.REQUIREMENTS_PEX),
-          self.context.products.get_data(GatherSources.PYTHON_SOURCES)
-        ]
+        source_pexes = self.context.products.get_data(GatherSources.PythonSources).all()
+        requirements_pex = self.context.products.get_data(ResolveRequirements.REQUIREMENTS_PEX)
+        pexes = [requirements_pex] + source_pexes
 
         if self.extra_requirements():
           extra_reqs = [PythonRequirement(req_str) for req_str in self.extra_requirements()]
diff --git a/src/python/pants/backend/python/tasks/setup_py.py b/src/python/pants/backend/python/tasks/setup_py.py
index a408a778cd4..ae0f0317f1b 100644
--- a/src/python/pants/backend/python/tasks/setup_py.py
+++ b/src/python/pants/backend/python/tasks/setup_py.py
@@ -22,7 +22,6 @@
 from pants.backend.python.targets.python_binary import PythonBinary
 from pants.backend.python.targets.python_requirement_library import PythonRequirementLibrary
 from pants.backend.python.targets.python_target import PythonTarget
-from pants.backend.python.tasks.gather_sources import GatherSources
 from pants.base.build_environment import get_buildroot
 from pants.base.exceptions import TargetDefinitionException, TaskError
 from pants.base.specs import SiblingAddresses
@@ -343,7 +342,6 @@ def dependencies(self, target):
 
   @classmethod
   def prepare(cls, options, round_manager):
-    round_manager.require_data(GatherSources.PYTHON_SOURCES)
     round_manager.require_data(PythonInterpreter)
 
   @classmethod
diff --git a/tests/python/pants_test/backend/python/tasks/test_gather_sources.py b/tests/python/pants_test/backend/python/tasks/test_gather_sources.py
index 5e6481298c4..a0e52c5f4bc 100644
--- a/tests/python/pants_test/backend/python/tasks/test_gather_sources.py
+++ b/tests/python/pants_test/backend/python/tasks/test_gather_sources.py
@@ -33,6 +33,7 @@ def setUp(self):
       'src/python/one/bar.py': 'bar_py_content',
       'src/python/two/baz.py': 'baz_py_content',
       'resources/qux/quux.txt': 'quux_txt_content',
+      'more/src/python/three/corge.py': 'corge_py_content',
     }
     # Pants does not do auto-detection of Resources target roots unless they are nested under some
     # other source root so we erect a manual resources root here.
@@ -41,32 +42,103 @@ def setUp(self):
     for rel_path, content in self.filemap.items():
       self.create_file(rel_path, content)
 
-    self.sources1 = self.make_target(spec='src/python/one:sources1_tgt', target_type=PythonLibrary,
-                                     sources=['foo.py', 'bar.py'])
-    self.sources2 = self.make_target(spec='src/python/two:sources2_tgt', target_type=PythonLibrary,
-                                     sources=['baz.py'])
-    self.resources = self.make_target(spec='resources/qux:resources_tgt', target_type=Resources,
+    self.resources = self.make_target(spec='resources/qux:resources_tgt',
+                                      target_type=Resources,
                                       sources=['quux.txt'])
-    self.files = self.make_target(spec='resources/qux:files_tgt', target_type=Files,
+    self.files = self.make_target(spec='resources/qux:files_tgt',
+                                  target_type=Files,
                                   sources=['quux.txt'])
-
-  def _assert_content(self, pex, relpath, prefix=None):
-    expected_content = self.filemap[os.path.join(prefix, relpath) if prefix else relpath]
-    with open(os.path.join(pex.path(), relpath)) as infile:
-      content = infile.read()
-    self.assertEquals(expected_content, content)
+    self.sources1 = self.make_target(spec='src/python/one:sources1_tgt',
+                                     target_type=PythonLibrary,
+                                     sources=['foo.py', 'bar.py'],
+                                     dependencies=[self.resources])
+    self.sources2 = self.make_target(spec='src/python/two:sources2_tgt',
+                                     target_type=PythonLibrary,
+                                     sources=['baz.py'],
+                                     dependencies=[self.files])
+    self.sources3 = self.make_target(spec='more/src/python/three:sources3_tgt',
+                                     target_type=PythonLibrary,
+                                     sources=['corge.py'],
+                                     dependencies=[self.files, self.resources])
+
+  def _assert_content(self, python_sources, target):
+    pex = python_sources.for_target(target)
+    self._assert_content_in_pex(pex, target)
+    return pex
+
+  def _extract_files(self, target):
+    if type(target) == Files:
+      to_filemap_key = lambda path: path
+      files = target.sources_relative_to_buildroot()
+    else:
+      to_filemap_key = lambda path: os.path.join(target.target_base, path)
+      files = target.sources_relative_to_source_root()
+    return to_filemap_key, files
+
+  def _assert_content_in_pex(self, pex, target):
+    to_filemap_key, files = self._extract_files(target)
+    pex_path = pex.path()
+    for path in files:
+      expected_content = self.filemap[to_filemap_key(path)]
+      with open(os.path.join(pex_path, path)) as infile:
+        content = infile.read()
+      self.assertEquals(expected_content, content)
+
+  def _assert_content_not_in_pex(self, pex, target):
+    _, files = self._extract_files(target)
+    pex_path = pex.path()
+    for path in files:
+      self.assertFalse(os.path.exists(os.path.join(pex_path, path)))
 
   def test_gather_sources(self):
-    pex = self._gather_sources([self.sources1, self.sources2, self.resources])
-    self._assert_content(pex, 'one/foo.py', prefix='src/python')
-    self._assert_content(pex, 'one/bar.py', prefix='src/python')
-    self._assert_content(pex, 'two/baz.py', prefix='src/python')
-    self._assert_content(pex, 'qux/quux.txt', prefix='resources')
+    python_sources = self._gather_sources([self.sources1,
+                                           # These files should not be gathered since they are not
+                                           # a dependency of any python targets in play.
+                                           self.files])
+    pex = self._assert_content(python_sources, self.sources1)
+    self._assert_content_in_pex(pex, self.resources)
+    self._assert_content_not_in_pex(pex, self.sources2)
+    self._assert_content_not_in_pex(pex, self.files)
 
   def test_gather_files(self):
-    pex = self._gather_sources([self.sources2, self.files])
-    self._assert_content(pex, 'two/baz.py', prefix='src/python')
-    self._assert_content(pex, 'resources/qux/quux.txt')
+    python_sources = self._gather_sources([self.sources2,
+                                           # These resources should not be gathered since they are
+                                           # not a dependency of any python targets in play.
+                                           self.resources])
+    pex = self._assert_content(python_sources, self.sources2)
+    self._assert_content_in_pex(pex, self.files)
+    self._assert_content_not_in_pex(pex, self.sources1)
+    self._assert_content_not_in_pex(pex, self.resources)
+
+  def test_gather_resources_into_multiple_pexes(self):
+    python_sources = self._gather_sources([self.sources1, self.sources2, self.sources3])
+
+    pex1 = self._assert_content(python_sources, self.sources1)
+    pex2 = self._assert_content(python_sources, self.sources2)
+    self.assertIs(pex1, pex2)  # sources1 and sources2 share the same source root.
+    self._assert_content_in_pex(pex1, self.files)
+    self._assert_content_in_pex(pex1, self.resources)
+    self._assert_content_not_in_pex(pex1, self.sources3)
+
+    pex3 = self._assert_content(python_sources, self.sources3)
+    self.assertIsNot(pex3, pex1)  # sources3 has a different source root from sources1 and sources2.
+    self._assert_content_in_pex(pex3, self.files)
+    self._assert_content_in_pex(pex3, self.resources)
+    self._assert_content_not_in_pex(pex3, self.sources1)
+    self._assert_content_not_in_pex(pex3, self.sources2)
+
+    self.assertEqual([pex1, pex3], python_sources.all())
+
+  def test_order_respected(self):
+    python_sources = self._gather_sources([self.sources1, self.sources3])
+    pex1 = python_sources.for_target(self.sources1)
+    pex3 = python_sources.for_target(self.sources3)
+    self.assertEqual([pex1, pex3], python_sources.all())
+
+    python_sources = self._gather_sources([self.sources3, self.sources1])
+    pex1 = python_sources.for_target(self.sources1)
+    pex3 = python_sources.for_target(self.sources3)
+    self.assertEqual([pex3, pex1], python_sources.all())
 
   def _gather_sources(self, target_roots):
     context = self.context(target_roots=target_roots, for_subsystems=[PythonSetup, PythonRepos])
@@ -84,4 +156,4 @@ def _gather_sources(self, target_roots):
     task = self.create_task(context)
     task.execute()
 
-    return context.products.get_data(GatherSources.PYTHON_SOURCES)
+    return context.products.get_data(GatherSources.PythonSources)
diff --git a/tests/python/pants_test/backend/python/tasks/test_pytest_run.py b/tests/python/pants_test/backend/python/tasks/test_pytest_run.py
index 89dd6aebf7a..866d1e591c8 100644
--- a/tests/python/pants_test/backend/python/tasks/test_pytest_run.py
+++ b/tests/python/pants_test/backend/python/tasks/test_pytest_run.py
@@ -10,6 +10,8 @@
 
 import coverage
 
+from pants.backend.python.targets.python_library import PythonLibrary
+from pants.backend.python.targets.python_tests import PythonTests
 from pants.backend.python.tasks.gather_sources import GatherSources
 from pants.backend.python.tasks.pytest_prep import PytestPrep
 from pants.backend.python.tasks.pytest_run import PytestResult, PytestRun
@@ -48,7 +50,6 @@ def try_run_tests(self, targets, *passthru_args, **options):
       return e.failed_targets
 
   def _prepare_test_run(self, targets, *passthru_args, **options):
-    self.reset_build_graph()
     test_options = {
       'colors': False,
       'level': 'info'  # When debugging a test failure it may be helpful to set this to 'debug'.
@@ -151,6 +152,7 @@ def setUp(self):
                                read_from=None,
                                write_to=None)
 
+    # Targets under test.
     self.create_file(
         'lib/core.py',
         dedent("""
@@ -161,21 +163,45 @@ def one():  # line 1
           def two():  # line 5
             return 2  # line 6
         """).strip())
-    self.add_to_build_file(
-        'lib',
+    core_lib = self.make_target(spec='lib:core',
+                                target_type=PythonLibrary,
+                                sources=['core.py'])
+
+    self.create_file(
+        'app/app.py',
+        dedent("""
+          import core          # line 1
+                               # line 2
+                               # line 3
+          def use_two():       # line 4
+            return core.two()  # line 5
+        """).strip())
+    app_lib = self.make_target(spec='app',
+                               target_type=PythonLibrary,
+                               sources=['app.py'],
+                               dependencies=[core_lib])
+
+    # Test targets.
+    self.create_file(
+        'tests/test_app.py',
         dedent("""
-          python_library(
-            name='core',
-            sources=[
-              'core.py'
-            ]
-          )
+          import unittest
+
+          import app
+
+          class AppTest(unittest.TestCase):
+            def test_use_two(self):
+              self.assertEqual(2, app.use_two())
         """))
+    self.app = self.make_target(spec='tests:app',
+                                target_type=PythonTests,
+                                sources=['test_app.py'],
+                                dependencies=[app_lib])
 
     self.create_file(
         'tests/test_core_green.py',
         dedent("""
-          import unittest2 as unittest
+          import unittest
 
           import core
 
@@ -183,10 +209,16 @@ class CoreGreenTest(unittest.TestCase):
             def test_one(self):
               self.assertEqual(1, core.one())
         """))
+    self.green = self.make_target(spec='tests:green',
+                                  target_type=PythonTests,
+                                  sources=['test_core_green.py'],
+                                  dependencies=[core_lib],
+                                  coverage=['core'])
+
     self.create_file(
-      'tests/test_core_green2.py',
-      dedent("""
-          import unittest2 as unittest
+        'tests/test_core_green2.py',
+        dedent("""
+          import unittest
 
           import core
 
@@ -194,10 +226,16 @@ class CoreGreen2Test(unittest.TestCase):
             def test_one(self):
               self.assertEqual(1, core.one())
         """))
+    self.green2 = self.make_target(spec='tests:green2',
+                                   target_type=PythonTests,
+                                   sources=['test_core_green2.py'],
+                                   dependencies=[core_lib],
+                                   coverage=['core'])
+
     self.create_file(
-      'tests/test_core_green3.py',
-      dedent("""
-          import unittest2 as unittest
+        'tests/test_core_green3.py',
+        dedent("""
+          import unittest
 
           import core
 
@@ -205,6 +243,12 @@ class CoreGreen3Test(unittest.TestCase):
             def test_one(self):
               self.assertEqual(1, core.one())
         """))
+    self.green3 = self.make_target(spec='tests:green3',
+                                   target_type=PythonTests,
+                                   sources=['test_core_green3.py'],
+                                   dependencies=[core_lib],
+                                   coverage=['core'])
+
     self.create_file(
         'tests/test_core_red.py',
         dedent("""
@@ -213,10 +257,16 @@ def test_one(self):
           def test_two():
             assert 1 == core.two()
         """))
+    self.red = self.make_target(spec='tests:red',
+                                target_type=PythonTests,
+                                sources=['test_core_red.py'],
+                                dependencies=[core_lib],
+                                coverage=['core'])
+
     self.create_file(
         'tests/test_core_red_in_class.py',
         dedent("""
-          import unittest2 as unittest
+          import unittest
 
           import core
 
@@ -224,195 +274,72 @@ class CoreRedClassTest(unittest.TestCase):
             def test_one_in_class(self):
               self.assertEqual(1, core.two())
         """))
+    self.red_in_class = self.make_target(spec='tests:red_in_class',
+                                         target_type=PythonTests,
+                                         sources=['test_core_red_in_class.py'],
+                                         dependencies=[core_lib],
+                                         coverage=['core'])
+
     self.create_file(
-      'tests/test_core_sleep.py',
-      dedent("""
+        'tests/test_core_sleep.py',
+        dedent("""
           import core
 
           def test_three():
             assert 1 == core.one()
         """))
-    self.create_file(
-      'tests/test_error.py',
-      dedent("""
-        def test_error(bad_fixture):
-          pass
-      """)
-    )
-    self.create_file(
-      'tests/test_failure_outside_function.py',
-      dedent("""
-      def null():
-        pass
+    self.sleep_no_timeout = self.make_target(spec='tests:sleep_no_timeout',
+                                             target_type=PythonTests,
+                                             sources=['test_core_sleep.py'],
+                                             dependencies=[core_lib],
+                                             coverage=['core'],
+                                             timeout=0)
+    self.sleep_timeout = self.make_target(spec='tests:sleep_timeout',
+                                          target_type=PythonTests,
+                                          sources=['test_core_sleep.py'],
+                                          dependencies=[core_lib],
+                                          coverage=['core'],
+                                          timeout=1)
 
-      assert(False)
-      """
-        )
-    )
     self.create_file(
-      'tests/conftest.py', self._CONFTEST_CONTENT
-    )
-
-    self.add_to_build_file(
-        'tests',
+        'tests/test_error.py',
         dedent("""
-          python_tests(
-            name='error',
-            sources=[
-              'test_error.py'
-            ],
-          )
-
-          python_tests(
-            name='failure_outside_function',
-            sources=[
-              'test_failure_outside_function.py',
-            ],
-          )
-
-          python_tests(
-            name='green',
-            sources=[
-              'test_core_green.py'
-            ],
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='green2',
-            sources=[
-              'test_core_green2.py'
-            ],
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='green3',
-            sources=[
-              'test_core_green3.py'
-            ],
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='red',
-            sources=[
-              'test_core_red.py',
-            ],
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='red_in_class',
-            sources=[
-              'test_core_red_in_class.py',
-            ],
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='sleep_no_timeout',
-            sources=[
-              'test_core_sleep.py',
-            ],
-            timeout = 0,
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='sleep_timeout',
-            sources=[
-              'test_core_sleep.py',
-            ],
-            timeout = 1,
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='all',
-            sources=[
-              'test_core_green.py',
-              'test_core_red.py',
-            ],
-            dependencies=[
-              'lib:core'
-            ]
-          )
-
-          python_tests(
-            name='all-with-coverage',
-            sources=[
-              'test_core_green.py',
-              'test_core_red.py'
-            ],
-            dependencies=[
-              'lib:core'
-            ],
-            coverage=[
-              'core'
-            ]
-          )
-
-          python_tests(
-            name='green-with-conftest',
-            sources=[
-              'conftest.py',
-              'test_core_green.py',
-            ],
-            dependencies=[
-              'lib:core',
-            ]
-          )
+          def test_error(bad_fixture):
+            pass
         """))
-    self.green = self.target('tests:green')
-    self.green2 = self.target('tests:green2')
-    self.green3 = self.target('tests:green3')
+    self.error = self.make_target(spec='tests:error',
+                                  target_type=PythonTests,
+                                  sources=['test_error.py'])
 
-    self.red = self.target('tests:red')
-    self.red_in_class = self.target('tests:red_in_class')
-    self.sleep_no_timeout = self.target('tests:sleep_no_timeout')
-    self.sleep_timeout = self.target('tests:sleep_timeout')
-    self.error = self.target('tests:error')
-    self.failure_outside_function = self.target('tests:failure_outside_function')
+    self.create_file(
+        'tests/test_failure_outside_function.py',
+        dedent("""
+        def null():
+          pass
 
-    self.all = self.target('tests:all')
-    self.all_with_coverage = self.target('tests:all-with-coverage')
-    self.green_with_conftest = self.target('tests:green-with-conftest')
+        assert(False)
+        """))
+    self.failure_outside_function = self.make_target(spec='tests:failure_outside_function',
+                                                     target_type=PythonTests,
+                                                     sources=['test_failure_outside_function.py'])
+
+    self.create_file('tests/conftest.py', self._CONFTEST_CONTENT)
+    self.green_with_conftest = self.make_target(spec='tests:green-with-conftest',
+                                                target_type=PythonTests,
+                                                sources=['conftest.py', 'test_core_green.py'],
+                                                dependencies=[core_lib])
+
+    self.all = self.make_target(spec='tests:all',
+                                target_type=PythonTests,
+                                sources=['test_core_green.py',
+                                         'test_core_red.py'],
+                                dependencies=[core_lib])
+
+    self.all_with_cov = self.make_target(spec='tests:all-with-coverage',
+                                         target_type=PythonTests,
+                                         sources=['test_core_green.py', 'test_core_red.py'],
+                                         dependencies=[core_lib],
+                                         coverage=['core'])
 
   @ensure_cached(PytestRun, expected_num_artifacts=0)
   def test_error(self):
@@ -505,49 +432,63 @@ def coverage_data_file(self):
 
   def load_coverage_data(self):
     path = os.path.join(self.build_root, 'lib', 'core.py')
+    return self.load_coverage_data_for(path)
+
+  def load_coverage_data_for(self, covered_path):
     data_file = self.coverage_data_file()
     self.assertTrue(os.path.isfile(data_file))
     coverage_data = coverage.coverage(data_file=data_file)
     coverage_data.load()
-    _, all_statements, not_run_statements, _ = coverage_data.analysis(path)
+    _, all_statements, not_run_statements, _ = coverage_data.analysis(covered_path)
     return all_statements, not_run_statements
 
-  @ensure_cached(PytestRun, expected_num_artifacts=1)
-  def test_coverage_auto_option(self):
-    simple_coverage_kwargs = {'coverage': 'auto'}
-
+  def run_coverage_auto(self, targets, failed_targets=None):
     self.assertFalse(os.path.isfile(self.coverage_data_file()))
+    simple_coverage_kwargs = {'coverage': 'auto'}
+    if failed_targets:
+      self.run_failing_tests(targets=targets,
+                             failed_targets=failed_targets,
+                             **simple_coverage_kwargs)
+    else:
+      self.run_tests(targets=targets, **simple_coverage_kwargs)
+    return self.load_coverage_data()
 
-    self.run_tests(targets=[self.green], **simple_coverage_kwargs)
-    all_statements, not_run_statements = self.load_coverage_data()
+  @ensure_cached(PytestRun, expected_num_artifacts=1)
+  def test_coverage_auto_option_green(self):
+    all_statements, not_run_statements = self.run_coverage_auto(targets=[self.green])
     self.assertEqual([1, 2, 5, 6], all_statements)
     self.assertEqual([6], not_run_statements)
 
-    self.run_failing_tests(targets=[self.red], failed_targets=[self.red], **simple_coverage_kwargs)
-    all_statements, not_run_statements = self.load_coverage_data()
+  @ensure_cached(PytestRun, expected_num_artifacts=0)
+  def test_coverage_auto_option_red(self):
+    all_statements, not_run_statements = self.run_coverage_auto(targets=[self.red],
+                                                                failed_targets=[self.red])
     self.assertEqual([1, 2, 5, 6], all_statements)
     self.assertEqual([2], not_run_statements)
 
-    self.run_failing_tests(targets=[self.green, self.red], failed_targets=[self.red],
-                           **simple_coverage_kwargs)
-    all_statements, not_run_statements = self.load_coverage_data()
+  @ensure_cached(PytestRun, expected_num_artifacts=0)
+  def test_coverage_auto_option_mixed_multiple_targets(self):
+    all_statements, not_run_statements = self.run_coverage_auto(targets=[self.green, self.red],
+                                                                failed_targets=[self.red])
     self.assertEqual([1, 2, 5, 6], all_statements)
-    # The green target run should be cached and thus not covered in this second run.
-    self.assertEqual([2], not_run_statements)
+    self.assertEqual([], not_run_statements)
+
+  @ensure_cached(PytestRun, expected_num_artifacts=0)
+  def test_coverage_auto_option_mixed_single_target(self):
+    all_statements, not_run_statements = self.run_coverage_auto(targets=[self.all_with_cov],
+                                                                failed_targets=[self.all_with_cov])
+    self.assertEqual([1, 2, 5, 6], all_statements)
+    self.assertEqual([], not_run_statements)
 
+  @ensure_cached(PytestRun, expected_num_artifacts=0)
+  def test_coverage_auto_option_no_explicit_coverage_idiosyncratic_layout(self):
     # The all target has no coverage attribute and the code under test does not follow the
     # auto-discover pattern so we should get no coverage.
-    self.run_failing_tests(targets=[self.all], failed_targets=[self.all], **simple_coverage_kwargs)
-    all_statements, not_run_statements = self.load_coverage_data()
+    all_statements, not_run_statements = self.run_coverage_auto(targets=[self.all],
+                                                                failed_targets=[self.all])
     self.assertEqual([1, 2, 5, 6], all_statements)
     self.assertEqual([1, 2, 5, 6], not_run_statements)
 
-    self.run_failing_tests(targets=[self.all_with_coverage],
-                           failed_targets=[self.all_with_coverage], **simple_coverage_kwargs)
-    all_statements, not_run_statements = self.load_coverage_data()
-    self.assertEqual([1, 2, 5, 6], all_statements)
-    self.assertEqual([], not_run_statements)
-
   @ensure_cached(PytestRun, expected_num_artifacts=0)
   def test_coverage_modules_dne_option(self):
     self.assertFalse(os.path.isfile(self.coverage_data_file()))
@@ -577,6 +518,44 @@ def test_coverage_paths_option(self):
     self.assertEqual([1, 2, 5, 6], all_statements)
     self.assertEqual([], not_run_statements)
 
+  @ensure_cached(PytestRun, expected_num_artifacts=1)
+  def test_coverage_issue_5314_primary_source_root(self):
+    self.assertFalse(os.path.isfile(self.coverage_data_file()))
+
+    self.run_tests(targets=[self.app], coverage='app')
+
+    app_path = os.path.join(self.build_root, 'app', 'app.py')
+    all_statements, not_run_statements = self.load_coverage_data_for(app_path)
+    self.assertEqual([1, 4, 5], all_statements)
+    self.assertEqual([], not_run_statements)
+
+  @ensure_cached(PytestRun, expected_num_artifacts=1)
+  def test_coverage_issue_5314_secondary_source_root(self):
+    self.assertFalse(os.path.isfile(self.coverage_data_file()))
+
+    self.run_tests(targets=[self.app], coverage='core')
+
+    core_path = os.path.join(self.build_root, 'lib', 'core.py')
+    all_statements, not_run_statements = self.load_coverage_data_for(core_path)
+    self.assertEqual([1, 2, 5, 6], all_statements)
+    self.assertEqual([2], not_run_statements)
+
+  @ensure_cached(PytestRun, expected_num_artifacts=1)
+  def test_coverage_issue_5314_all_source_roots(self):
+    self.assertFalse(os.path.isfile(self.coverage_data_file()))
+
+    self.run_tests(targets=[self.app], coverage='app,core')
+
+    app_path = os.path.join(self.build_root, 'app', 'app.py')
+    all_statements, not_run_statements = self.load_coverage_data_for(app_path)
+    self.assertEqual([1, 4, 5], all_statements)
+    self.assertEqual([], not_run_statements)
+
+    core_path = os.path.join(self.build_root, 'lib', 'core.py')
+    all_statements, not_run_statements = self.load_coverage_data_for(core_path)
+    self.assertEqual([1, 2, 5, 6], all_statements)
+    self.assertEqual([2], not_run_statements)
+
   @ensure_cached(PytestRun, expected_num_artifacts=1)
   def test_sharding(self):
     shard0_failed_targets = self.try_run_tests(targets=[self.red, self.green], test_shard='0/2')