Merge branch 'raoul/440-source-cache-remotes' into 'master'

Remote source cache

Closes #440

See merge request BuildStream/buildstream!1214

26 files changed, 956 insertions, 78 deletions

diff --git a/buildstream/_artifactcache.py b/buildstream/_artifactcache.py
index f97da4661..3ca6c6e60 100644
--- a/buildstream/_artifactcache.py
+++ b/buildstream/_artifactcache.py
@@ -263,48 +263,6 @@ class ArtifactCache(BaseCache):
 
         return self.cas.diff(ref_a, ref_b, subdir=subdir)
 
-    # has_fetch_remotes():
-    #
-    # Check whether any remote repositories are available for fetching.
-    #
-    # Args:
-    #     element (Element): The Element to check
-    #
-    # Returns: True if any remote repositories are configured, False otherwise
-    #
-    def has_fetch_remotes(self, *, element=None):
-        if not self._has_fetch_remotes:
-            # No project has fetch remotes
-            return False
-        elif element is None:
-            # At least one (sub)project has fetch remotes
-            return True
-        else:
-            # Check whether the specified element's project has fetch remotes
-            remotes_for_project = self._remotes[element._get_project()]
-            return bool(remotes_for_project)
-
-    # has_push_remotes():
-    #
-    # Check whether any remote repositories are available for pushing.
-    #
-    # Args:
-    #     element (Element): The Element to check
-    #
-    # Returns: True if any remote repository is configured, False otherwise
-    #
-    def has_push_remotes(self, *, element=None):
-        if not self._has_push_remotes:
-            # No project has push remotes
-            return False
-        elif element is None:
-            # At least one (sub)project has push remotes
-            return True
-        else:
-            # Check whether the specified element's project has push remotes
-            remotes_for_project = self._remotes[element._get_project()]
-            return any(remote.spec.push for remote in remotes_for_project)
-
     # push():
     #
     # Push committed artifact to remote repository.
@@ -337,7 +295,7 @@ class ArtifactCache(BaseCache):
                 element.info("Pushed artifact {} -> {}".format(display_key, remote.spec.url))
                 pushed = True
             else:
-                element.info("Remote ({}) already has {} cached".format(
+                element.info("Remote ({}) already has artifact {} cached".format(
                     remote.spec.url, element._get_brief_display_key()
                 ))
 
@@ -372,7 +330,7 @@ class ArtifactCache(BaseCache):
                     # no need to pull from additional remotes
                     return True
                 else:
-                    element.info("Remote ({}) does not have {} cached".format(
+                    element.info("Remote ({}) does not have artifact {} cached".format(
                         remote.spec.url, element._get_brief_display_key()
                     ))
 
diff --git a/buildstream/_basecache.py b/buildstream/_basecache.py
index a8c58e48f..696cbf9c1 100644
--- a/buildstream/_basecache.py
+++ b/buildstream/_basecache.py
@@ -190,6 +190,48 @@ class BaseCache():
 
             self._remotes[project] = project_remotes
 
+    # has_fetch_remotes():
+    #
+    # Check whether any remote repositories are available for fetching.
+    #
+    # Args:
+    #     plugin (Plugin): The Plugin to check
+    #
+    # Returns: True if any remote repositories are configured, False otherwise
+    #
+    def has_fetch_remotes(self, *, plugin=None):
+        if not self._has_fetch_remotes:
+            # No project has fetch remotes
+            return False
+        elif plugin is None:
+            # At least one (sub)project has fetch remotes
+            return True
+        else:
+            # Check whether the specified element's project has fetch remotes
+            remotes_for_project = self._remotes[plugin._get_project()]
+            return bool(remotes_for_project)
+
+    # has_push_remotes():
+    #
+    # Check whether any remote repositories are available for pushing.
+    #
+    # Args:
+    #     element (Element): The Element to check
+    #
+    # Returns: True if any remote repository is configured, False otherwise
+    #
+    def has_push_remotes(self, *, plugin=None):
+        if not self._has_push_remotes:
+            # No project has push remotes
+            return False
+        elif plugin is None:
+            # At least one (sub)project has push remotes
+            return True
+        else:
+            # Check whether the specified element's project has push remotes
+            remotes_for_project = self._remotes[plugin._get_project()]
+            return any(remote.spec.push for remote in remotes_for_project)
+
     ################################################
     #               Local Private Methods          #
     ################################################
diff --git a/buildstream/_frontend/cli.py b/buildstream/_frontend/cli.py
index 25298a684..a1a780cc4 100644
--- a/buildstream/_frontend/cli.py
+++ b/buildstream/_frontend/cli.py
@@ -622,10 +622,12 @@ def source():
               help="Track new source references before fetching")
 @click.option('--track-cross-junctions', '-J', default=False, is_flag=True,
               help="Allow tracking to cross junction boundaries")
+@click.option('--remote', '-r', default=None,
+              help="The URL of the remote source cache (defaults to the first configured cache)")
 @click.argument('elements', nargs=-1,
                 type=click.Path(readable=False))
 @click.pass_obj
-def source_fetch(app, elements, deps, track_, except_, track_cross_junctions):
+def source_fetch(app, elements, deps, track_, except_, track_cross_junctions, remote):
     """Fetch sources required to build the pipeline
 
     Specifying no elements will result in fetching the default targets
@@ -666,7 +668,8 @@ def source_fetch(app, elements, deps, track_, except_, track_cross_junctions):
                          selection=deps,
                          except_targets=except_,
                          track_targets=track_,
-                         track_cross_junctions=track_cross_junctions)
+                         track_cross_junctions=track_cross_junctions,
+                         remote=remote)
 
 
 ##################################################################
diff --git a/buildstream/_frontend/widget.py b/buildstream/_frontend/widget.py
index ef31b8ba7..f092cb5ec 100644
--- a/buildstream/_frontend/widget.py
+++ b/buildstream/_frontend/widget.py
@@ -201,7 +201,7 @@ class ElementName(Widget):
         if not action_name:
             action_name = "Main"
 
-        return self.content_profile.fmt("{: >5}".format(action_name.lower())) + \
+        return self.content_profile.fmt("{: >8}".format(action_name.lower())) + \
             self.format_profile.fmt(':') + self.content_profile.fmt(name)
 
 
diff --git a/buildstream/_scheduler/__init__.py b/buildstream/_scheduler/__init__.py
index 470859864..d2f458fa5 100644
--- a/buildstream/_scheduler/__init__.py
+++ b/buildstream/_scheduler/__init__.py
@@ -20,9 +20,10 @@
 from .queues import Queue, QueueStatus
 
 from .queues.fetchqueue import FetchQueue
+from .queues.sourcepushqueue import SourcePushQueue
 from .queues.trackqueue import TrackQueue
 from .queues.buildqueue import BuildQueue
-from .queues.pushqueue import PushQueue
+from .queues.artifactpushqueue import ArtifactPushQueue
 from .queues.pullqueue import PullQueue
 
 from .scheduler import Scheduler, SchedStatus
diff --git a/buildstream/_scheduler/queues/pushqueue.py b/buildstream/_scheduler/queues/artifactpushqueue.py
index 35532d23d..b861d4fc7 100644
--- a/buildstream/_scheduler/queues/pushqueue.py
+++ b/buildstream/_scheduler/queues/artifactpushqueue.py
@@ -26,7 +26,7 @@ from ..._exceptions import SkipJob
 
 # A queue which pushes element artifacts
 #
-class PushQueue(Queue):
+class ArtifactPushQueue(Queue):
 
     action_name = "Push"
     complete_name = "Pushed"
diff --git a/buildstream/_scheduler/queues/fetchqueue.py b/buildstream/_scheduler/queues/fetchqueue.py
index 546c65b65..9edeebb1d 100644
--- a/buildstream/_scheduler/queues/fetchqueue.py
+++ b/buildstream/_scheduler/queues/fetchqueue.py
@@ -73,5 +73,8 @@ class FetchQueue(Queue):
 
         element._fetch_done()
 
-        # Successful fetch, we must be CACHED now
-        assert element._get_consistency() == Consistency.CACHED
+        # Successful fetch, we must be CACHED or in the sourcecache
+        if self._fetch_original:
+            assert element._get_consistency() == Consistency.CACHED
+        else:
+            assert element._source_cached()
diff --git a/buildstream/_scheduler/queues/sourcepushqueue.py b/buildstream/_scheduler/queues/sourcepushqueue.py
new file mode 100644
index 000000000..c38460e6a
--- /dev/null
+++ b/buildstream/_scheduler/queues/sourcepushqueue.py
@@ -0,0 +1,42 @@
+#
+#  Copyright (C) 2019 Bloomberg Finance LP
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Authors:
+#        Raoul Hidalgo Charman <raoul.hidalgocharman@codethink.co.uk>
+
+from . import Queue, QueueStatus
+from ..resources import ResourceType
+from ..._exceptions import SkipJob
+
+
+# A queue which pushes staged sources
+#
+class SourcePushQueue(Queue):
+
+    action_name = "Src-push"
+    complete_name = "Sources pushed"
+    resources = [ResourceType.UPLOAD]
+
+    def process(self, element):
+        # Returns whether a source was pushed or not
+        if not element._source_push():
+            raise SkipJob(self.action_name)
+
+    def status(self, element):
+        if element._skip_source_push():
+            return QueueStatus.SKIP
+
+        return QueueStatus.READY
diff --git a/buildstream/_sourcecache.py b/buildstream/_sourcecache.py
index b21edaa81..219ecb1e9 100644
--- a/buildstream/_sourcecache.py
+++ b/buildstream/_sourcecache.py
@@ -20,7 +20,7 @@
 from ._cas import CASRemoteSpec
 from .storage._casbaseddirectory import CasBasedDirectory
 from ._basecache import BaseCache
-from ._exceptions import CASCacheError, SourceCacheError
+from ._exceptions import CASError, CASCacheError, SourceCacheError
 from . import utils
 
 
@@ -143,3 +143,71 @@ class SourceCache(BaseCache):
             raise SourceCacheError("Error exporting source: {}".format(e))
 
         return CasBasedDirectory(self.cas, digest=digest)
+
+    # pull()
+    #
+    # Attempts to pull sources from configure remote source caches.
+    #
+    # Args:
+    #    source (Source): The source we want to fetch
+    #    progress (callable|None): The progress callback
+    #
+    # Returns:
+    #    (bool): True if pull successful, False if not
+    def pull(self, source, *, progress=None):
+        ref = source._get_source_name()
+
+        project = source._get_project()
+
+        display_key = source._get_brief_display_key()
+
+        for remote in self._remotes[project]:
+            try:
+                source.status("Pulling source {} <- {}".format(display_key, remote.spec.url))
+
+                if self.cas.pull(ref, remote, progress=progress):
+                    source.info("Pulled source {} <- {}".format(display_key, remote.spec.url))
+                    # no need to pull from additional remotes
+                    return True
+                else:
+                    source.info("Remote ({}) does not have source {} cached".format(
+                        remote.spec.url, display_key))
+            except CASError as e:
+                raise SourceCacheError("Failed to pull source {}: {}".format(
+                    display_key, e)) from e
+        return False
+
+    # push()
+    #
+    # Push a source to configured remote source caches
+    #
+    # Args:
+    #    source (Source): source to push
+    #
+    # Returns:
+    #    (Bool): whether it pushed to a remote source cache
+    #
+    def push(self, source):
+        ref = source._get_source_name()
+        project = source._get_project()
+
+        # find configured push remotes for this source
+        if self._has_push_remotes:
+            push_remotes = [r for r in self._remotes[project] if r.spec.push]
+        else:
+            push_remotes = []
+
+        pushed = False
+
+        display_key = source._get_brief_display_key()
+        for remote in push_remotes:
+            remote.init()
+            source.status("Pushing source {} -> {}".format(display_key, remote.spec.url))
+            if self.cas.push([ref], remote):
+                source.info("Pushed source {} -> {}".format(display_key, remote.spec.url))
+                pushed = True
+            else:
+                source.info("Remote ({}) already has source {} cached"
+                            .format(remote.spec.url, display_key))
+
+        return pushed
diff --git a/buildstream/_stream.py b/buildstream/_stream.py
index d77b8d33b..d2ece163d 100644
--- a/buildstream/_stream.py
+++ b/buildstream/_stream.py
@@ -34,7 +34,8 @@ from fnmatch import fnmatch
 from ._artifactelement import verify_artifact_ref
 from ._exceptions import StreamError, ImplError, BstError, ArtifactElementError, CASCacheError
 from ._message import Message, MessageType
-from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, BuildQueue, PullQueue, PushQueue
+from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, \
+    SourcePushQueue, BuildQueue, PullQueue, ArtifactPushQueue
 from ._pipeline import Pipeline, PipelineSelection
 from ._profile import Topics, profile_start, profile_end
 from .types import _KeyStrength
@@ -77,6 +78,7 @@ class Stream():
         # Private members
         #
         self._artifacts = context.artifactcache
+        self._sourcecache = context.sourcecache
         self._context = context
         self._project = project
         self._pipeline = Pipeline(context, project, self._artifacts)
@@ -106,7 +108,7 @@ class Stream():
     #    targets (list of str): Targets to pull
     #    selection (PipelineSelection): The selection mode for the specified targets
     #    except_targets (list of str): Specified targets to except from fetching
-    #    use_artifact_config (bool): If artifact remote config should be loaded
+    #    use_artifact_config (bool): If artifact remote configs should be loaded
     #
     # Returns:
     #    (list of Element): The selected elements
@@ -239,6 +241,7 @@ class Stream():
                        ignore_junction_targets=ignore_junction_targets,
                        use_artifact_config=use_config,
                        artifact_remote_url=remote,
+                       use_source_config=True,
                        fetch_subprojects=True,
                        dynamic_plan=True)
 
@@ -259,10 +262,14 @@ class Stream():
             self._add_queue(PullQueue(self._scheduler))
 
         self._add_queue(FetchQueue(self._scheduler, skip_cached=True))
+
         self._add_queue(BuildQueue(self._scheduler))
 
         if self._artifacts.has_push_remotes():
-            self._add_queue(PushQueue(self._scheduler))
+            self._add_queue(ArtifactPushQueue(self._scheduler))
+
+        if self._sourcecache.has_push_remotes():
+            self._add_queue(SourcePushQueue(self._scheduler))
 
         # Enqueue elements
         #
@@ -281,12 +288,14 @@ class Stream():
     #    except_targets (list of str): Specified targets to except from fetching
     #    track_targets (bool): Whether to track selected targets in addition to fetching
     #    track_cross_junctions (bool): Whether tracking should cross junction boundaries
+    #    remote (str|None): The URL of a specific remote server to pull from.
     #
     def fetch(self, targets, *,
               selection=PipelineSelection.PLAN,
               except_targets=None,
               track_targets=False,
-              track_cross_junctions=False):
+              track_cross_junctions=False,
+              remote=None):
 
         if track_targets:
             track_targets = targets
@@ -297,13 +306,19 @@ class Stream():
             track_selection = PipelineSelection.NONE
             track_except_targets = ()
 
+        use_source_config = True
+        if remote:
+            use_source_config = False
+
         elements, track_elements = \
             self._load(targets, track_targets,
                        selection=selection, track_selection=track_selection,
                        except_targets=except_targets,
                        track_except_targets=track_except_targets,
                        track_cross_junctions=track_cross_junctions,
-                       fetch_subprojects=True)
+                       fetch_subprojects=True,
+                       use_source_config=use_source_config,
+                       source_remote_url=remote)
 
         # Delegated to a shared fetch method
         self._fetch(elements, track_elements=track_elements)
@@ -424,7 +439,7 @@ class Stream():
             self._add_queue(PullQueue(self._scheduler))
             self._enqueue_plan(require_buildtrees)
 
-        push_queue = PushQueue(self._scheduler)
+        push_queue = ArtifactPushQueue(self._scheduler)
         self._add_queue(push_queue)
         self._enqueue_plan(elements, queue=push_queue)
         self._run()
@@ -986,7 +1001,9 @@ class Stream():
     #    track_cross_junctions (bool): Whether tracking should cross junction boundaries
     #    ignore_junction_targets (bool): Whether junction targets should be filtered out
     #    use_artifact_config (bool): Whether to initialize artifacts with the config
-    #    artifact_remote_url (bool): A remote url for initializing the artifacts
+    #    use_source_config (bool): Whether to initialize remote source caches with the config
+    #    artifact_remote_url (str): A remote url for initializing the artifacts
+    #    source_remote_url (str): A remote url for initializing source caches
     #    fetch_subprojects (bool): Whether to fetch subprojects while loading
     #
     # Returns:
@@ -1002,7 +1019,9 @@ class Stream():
               track_cross_junctions=False,
               ignore_junction_targets=False,
               use_artifact_config=False,
+              use_source_config=False,
               artifact_remote_url=None,
+              source_remote_url=None,
               fetch_subprojects=False,
               dynamic_plan=False,
               load_refs=False):
@@ -1087,6 +1106,7 @@ class Stream():
 
         # Connect to remote caches, this needs to be done before resolving element state
         self._artifacts.setup_remotes(use_config=use_artifact_config, remote_url=artifact_remote_url)
+        self._sourcecache.setup_remotes(use_config=use_source_config, remote_url=source_remote_url)
 
         # Now move on to loading primary selection.
         #
@@ -1106,7 +1126,7 @@ class Stream():
         required_elements = functools.partial(self._pipeline.dependencies, elements, Scope.ALL)
         self._artifacts.mark_required_elements(required_elements())
 
-        self._context.sourcecache.mark_required_sources(
+        self._sourcecache.mark_required_sources(
             itertools.chain.from_iterable(
                 [element.sources() for element in required_elements()]))
 
@@ -1217,6 +1237,7 @@ class Stream():
 
         if track_elements:
             self._enqueue_plan(track_elements, queue=track_queue)
+
         self._enqueue_plan(fetch_plan)
         self._run()
 
diff --git a/buildstream/element.py b/buildstream/element.py
index cb04a9c15..b27f3e7df 100644
--- a/buildstream/element.py
+++ b/buildstream/element.py
@@ -213,6 +213,7 @@ class Element(Plugin):
         self.__weak_cache_key = None            # Our cached weak cache key
         self.__strict_cache_key = None          # Our cached cache key for strict builds
         self.__artifacts = context.artifactcache  # Artifact cache
+        self.__sourcecache = context.sourcecache  # Source cache
         self.__consistency = Consistency.INCONSISTENT  # Cached overall consistency state
         self.__strong_cached = None             # Whether we have a cached artifact
         self.__weak_cached = None               # Whether we have a cached artifact
@@ -1810,7 +1811,7 @@ class Element(Plugin):
 
         # Pull is pending if artifact remote server available
         # and pull has not been attempted yet
-        return self.__artifacts.has_fetch_remotes(element=self) and not self.__pull_done
+        return self.__artifacts.has_fetch_remotes(plugin=self) and not self.__pull_done
 
     # _pull_done()
     #
@@ -1855,6 +1856,25 @@ class Element(Plugin):
         # Notify successfull download
         return True
 
+    def _skip_source_push(self):
+        if not self.__sources or self._get_workspace():
+            return True
+        return not (self.__sourcecache.has_push_remotes(plugin=self) and
+                    self._source_cached())
+
+    def _source_push(self):
+        # try and push sources if we've got them
+        if self.__sourcecache.has_push_remotes(plugin=self) and self._source_cached():
+            sources = list(self.sources())
+            if sources:
+                source_pushed = self.__sourcecache.push(sources[-1])
+
+                if not source_pushed:
+                    return False
+
+        # Notify successful upload
+        return True
+
     # _skip_push():
     #
     # Determine whether we should create a push job for this element.
@@ -1863,7 +1883,7 @@ class Element(Plugin):
     #   (bool): True if this element does not need a push job to be created
     #
     def _skip_push(self):
-        if not self.__artifacts.has_push_remotes(element=self):
+        if not self.__artifacts.has_push_remotes(plugin=self):
             # No push remotes for this element's project
             return True
 
@@ -2111,16 +2131,20 @@ class Element(Plugin):
     #
     def _fetch(self, fetch_original=False):
         previous_sources = []
-        source = None
-        sourcecache = self._get_context().sourcecache
-
-        # check whether the final source is cached
-        for source in self.sources():
-            pass
+        sources = self.__sources
+        if sources and not fetch_original:
+            source = sources[-1]
+            if self.__sourcecache.contains(source):
+                return
 
-        if source and not fetch_original and sourcecache.contains(source):
-            return
+            # try and fetch from source cache
+            if source._get_consistency() < Consistency.CACHED and \
+                    self.__sourcecache.has_fetch_remotes() and \
+                    not self.__sourcecache.contains(source):
+                if self.__sourcecache.pull(source):
+                    return
 
+        # We need to fetch original sources
         for source in self.sources():
             source_consistency = source._get_consistency()
             if source_consistency != Consistency.CACHED:
diff --git a/buildstream/plugintestutils/runcli.py b/buildstream/plugintestutils/runcli.py
index c08dd0ff3..71d4b4039 100644
--- a/buildstream/plugintestutils/runcli.py
+++ b/buildstream/plugintestutils/runcli.py
@@ -225,7 +225,7 @@ class Result():
     #    (list): A list of element names
     #
     def get_tracked_elements(self):
-        tracked = re.findall(r'\[track:(\S+)\s*]', self.stderr)
+        tracked = re.findall(r'\[\s*track:(\S+)\s*]', self.stderr)
         if tracked is None:
             return []
 
diff --git a/buildstream/source.py b/buildstream/source.py
index af89ff8aa..9e1a8ef3e 100644
--- a/buildstream/source.py
+++ b/buildstream/source.py
@@ -986,6 +986,13 @@ class Source(Plugin):
             self.get_kind(),
             self._key)
 
+    def _get_brief_display_key(self):
+        context = self._get_context()
+        key = self._key
+
+        length = min(len(key), context.log_key_length)
+        return key[:length]
+
     #############################################################
     #                   Local Private Methods                   #
     #############################################################
diff --git a/doc/source/format_project.rst b/doc/source/format_project.rst
index a2216100f..f9d999a19 100644
--- a/doc/source/format_project.rst
+++ b/doc/source/format_project.rst
@@ -218,6 +218,33 @@ The use of ports are required to distinguish between pull only access and
 push/pull access. For information regarding the server/client certificates
 and keys, please see: :ref:`Key pair for the server <server_authentication>`.
 
+.. _project_source_cache:
+
+Source cache server
+~~~~~~~~~~~~~~~~~~~
+Exactly the same as artifact servers, source cache servers can be specified.
+
+.. code:: yaml
+
+  #
+  # Source caches
+  #
+  source-caches:
+    # A remote cache from which to download prestaged sources
+    - url: https://foo.com:11001
+      server.cert: server.crt
+    # A remote cache from which to upload/download prestaged sources
+    - url: https://foo.com:11002
+      server-cert: server.crt
+      client-cert: client.crt
+      client-key: client.key
+
+.. note::
+
+   As artifact caches work in exactly the same way, a configured artifact server
+   can also be used as a source cache server. If you want to use a server as
+   both you can put it under both artifacts and source caches configs.
+
 .. _project_remote_execution:
 
 Remote execution
diff --git a/doc/source/using_config.rst b/doc/source/using_config.rst
index 40b763e78..58ef160c1 100644
--- a/doc/source/using_config.rst
+++ b/doc/source/using_config.rst
@@ -100,6 +100,52 @@ pull only access and push/pull access. For information regarding this and the
 server/client certificates and keys, please see:
 :ref:`Key pair for the server <server_authentication>`.
 
+Source cache server
+~~~~~~~~~~~~~~~~~~~
+Similarly global and project specific source caches servers can be specified in
+the user configuration.
+
+1. Global source caches
+
+.. code:: yaml
+
+   #
+   # Source caches
+   #
+   source-caches:
+     # Add a cache to pull from
+     - url: https://cache.com/sources:11001
+       server-cert: server.crt
+     # Add a cache to push/pull to/from
+     - url: https://cache.com/sources:11002
+       server-cert: server.crt
+       client-cert: client.crt
+       client-key: client.key
+       push: true
+     # Add another cache to pull from
+     - url: https://anothercache.com/sources:8080
+       server-cert: another_server.crt
+
+2. Project specific source caches
+
+.. code:: yaml
+
+   projects:
+     project-name:
+       artifacts:
+         # Add a cache to pull from
+         - url: https://cache.com/sources:11001
+           server-cert: server.crt
+         # Add a cache to push/pull to/from
+         - url: https://cache.com/sources:11002
+           server-cert: server.crt
+           client-cert: client.crt
+           client-key: client.key
+           push: true
+         # Add another cache to pull from
+         - url: https://ourprojectcache.com/sources:8080
+           server-cert: project_server.crt
+
 .. _user_config_remote_execution:
 
 Remote execution
diff --git a/tests/artifactcache/expiry.py b/tests/artifactcache/expiry.py
index 38c0e21f0..7ada656ab 100644
--- a/tests/artifactcache/expiry.py
+++ b/tests/artifactcache/expiry.py
@@ -435,10 +435,10 @@ def test_cleanup_first(cli, datafiles):
     res = cli.run(project=project, args=['build', 'target2.bst'])
     res.assert_success()
 
-    # Find all of the activity (like push, pull, fetch) lines
+    # Find all of the activity (like push, pull, src-pull) lines
     results = re.findall(r'\[.*\]\[.*\]\[\s*(\S+):.*\]\s*START\s*.*\.log', res.stderr)
 
-    # Don't bother checking the order of 'fetch', it is allowed to start
+    # Don't bother checking the order of 'src-pull', it is allowed to start
     # before or after the initial cache size job, runs in parallel, and does
     # not require ResourceType.CACHE.
     results.remove('fetch')
diff --git a/tests/artifactcache/pull.py b/tests/artifactcache/pull.py
index 3e05bcecf..3c10c256c 100644
--- a/tests/artifactcache/pull.py
+++ b/tests/artifactcache/pull.py
@@ -150,7 +150,7 @@ def _test_pull(user_config_file, project_dir, cache_dir,
     # Manually setup the CAS remote
     cas.setup_remotes(use_config=True)
 
-    if cas.has_push_remotes(element=element):
+    if cas.has_push_remotes(plugin=element):
         # Push the element's artifact
         if not cas.pull(element, element_key):
             queue.put("Pull operation failed")
diff --git a/tests/artifactcache/push.py b/tests/artifactcache/push.py
index 18d6d80bc..69f3fbfbb 100644
--- a/tests/artifactcache/push.py
+++ b/tests/artifactcache/push.py
@@ -125,7 +125,7 @@ def _test_push(user_config_file, project_dir, element_name, element_key, queue):
     cas.setup_remotes(use_config=True)
     cas.initialize_remotes()
 
-    if cas.has_push_remotes(element=element):
+    if cas.has_push_remotes(plugin=element):
         # Push the element's artifact
         if not cas.push(element, [element_key]):
             queue.put("Push operation failed")
@@ -184,7 +184,7 @@ def test_push_directory(cli, tmpdir, datafiles):
         # Manually setup the CAS remote
         artifactcache.setup_remotes(use_config=True)
         artifactcache.initialize_remotes()
-        assert artifactcache.has_push_remotes(element=element)
+        assert artifactcache.has_push_remotes(plugin=element)
 
         # Recreate the CasBasedDirectory object from the cached artifact
         artifact_ref = element.get_artifact_name(element_key)
diff --git a/tests/frontend/buildtrack.py b/tests/frontend/buildtrack.py
index 2d8d0e383..5a3781dc6 100644
--- a/tests/frontend/buildtrack.py
+++ b/tests/frontend/buildtrack.py
@@ -303,11 +303,11 @@ def test_build_track_track_first(cli, datafiles, tmpdir, strict):
 
     # Assert that 1.bst successfully tracks before 0.bst builds
     track_messages = re.finditer(r'\[track:1.bst\s*]', result.stderr)
-    build_0 = re.search(r'\[build:0.bst\s*] START', result.stderr).start()
+    build_0 = re.search(r'\[\s*build:0.bst\s*] START', result.stderr).start()
     assert all(track_message.start() < build_0 for track_message in track_messages)
 
     # Assert that 2.bst is *only* rebuilt if we are in strict mode
-    build_2 = re.search(r'\[build:2.bst\s*] START', result.stderr)
+    build_2 = re.search(r'\[\s*build:2.bst\s*] START', result.stderr)
     if strict == '--strict':
         assert build_2 is not None
     else:
diff --git a/tests/frontend/remote-caches.py b/tests/frontend/remote-caches.py
new file mode 100644
index 000000000..3e3e226fc
--- /dev/null
+++ b/tests/frontend/remote-caches.py
@@ -0,0 +1,91 @@
+#
+#  Copyright (C) 2019 Bloomberg Finance LP
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Authors:
+#        Raoul Hidalgo Charman <raoul.hidalgocharman@codethink.co.uk>
+#
+# Pylint doesn't play well with fixtures and dependency injection from pytest
+# pylint: disable=redefined-outer-name
+import os
+import shutil
+import pytest
+
+from buildstream.plugintestutils import cli  # pylint: disable=unused-import
+from buildstream import _yaml
+
+from tests.testutils import create_artifact_share, create_element_size
+
+DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'project')
+
+
+def message_handler(message, context):
+    pass
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_source_artifact_caches(cli, tmpdir, datafiles):
+    cachedir = os.path.join(str(tmpdir), 'cache')
+    project_dir = str(datafiles)
+    element_path = os.path.join(project_dir, 'elements')
+
+    with create_artifact_share(os.path.join(str(tmpdir), 'share')) as share:
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+                'push': True,
+            },
+            'artifacts': {
+                'url': share.repo,
+                'push': True,
+            },
+            'cachedir': cachedir
+        }
+        _yaml.dump(_yaml.node_sanitize(user_config), filename=user_config_file)
+        cli.configure(user_config)
+
+        create_element_size('repo.bst', project_dir, element_path, [], 10000)
+
+        res = cli.run(project=project_dir, args=['build', 'repo.bst'])
+        res.assert_success()
+        assert "Pushed source " in res.stderr
+        assert "Pushed artifact " in res.stderr
+
+        # delete local sources and artifacts and check it pulls them
+        shutil.rmtree(os.path.join(cachedir, 'cas'))
+        shutil.rmtree(os.path.join(cachedir, 'sources'))
+
+        # this should just fetch the artifacts
+        res = cli.run(project=project_dir, args=['build', 'repo.bst'])
+        res.assert_success()
+        assert "Pulled artifact " in res.stderr
+        assert "Pulled source " not in res.stderr
+
+        # remove the artifact from the repo and check it pulls sources, builds
+        # and then pushes the artifacts
+        shutil.rmtree(os.path.join(cachedir, 'cas'))
+        print(os.listdir(os.path.join(share.repodir, 'cas', 'refs', 'heads')))
+        shutil.rmtree(os.path.join(share.repodir, 'cas', 'refs', 'heads', 'test'))
+
+        res = cli.run(project=project_dir, args=['build', 'repo.bst'])
+        res.assert_success()
+        assert "Remote ({}) does not have artifact ".format(share.repo) in res.stderr
+        assert "Pulled source" in res.stderr
+        assert "Caching artifact" in res.stderr
+        assert "Pushed artifact" in res.stderr
diff --git a/tests/sourcecache/fetch.py b/tests/sourcecache/fetch.py
new file mode 100644
index 000000000..86d2138c0
--- /dev/null
+++ b/tests/sourcecache/fetch.py
@@ -0,0 +1,222 @@
+#
+#  Copyright (C) 2019 Bloomberg Finance LP
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Authors:
+#        Raoul Hidalgo Charman <raoul.hidalgocharman@codethink.co.uk>
+#
+# Pylint doesn't play well with fixtures and dependency injection from pytest
+# pylint: disable=redefined-outer-name
+import os
+import shutil
+import pytest
+
+from buildstream._exceptions import ErrorDomain
+from buildstream._context import Context
+from buildstream._project import Project
+from buildstream import _yaml
+from buildstream.plugintestutils import cli  # pylint: disable=unused-import
+
+from tests.testutils import create_artifact_share, create_repo
+
+DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "project")
+
+
+def message_handler(message, context):
+    pass
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_source_fetch(cli, tmpdir, datafiles):
+    project_dir = str(datafiles)
+
+    # use artifact cache for sources for now, they should work the same
+    with create_artifact_share(os.path.join(str(tmpdir), 'sourceshare')) as share:
+        # configure using this share
+        cache_dir = os.path.join(str(tmpdir), 'cache')
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+            },
+            'cachedir': cache_dir,
+        }
+        _yaml.dump(_yaml.node_sanitize(user_config), filename=user_config_file)
+        cli.configure(user_config)
+
+        repo = create_repo('git', str(tmpdir))
+        ref = repo.create(os.path.join(project_dir, 'files'))
+        element_path = os.path.join(project_dir, 'elements')
+        element_name = 'fetch.bst'
+        element = {
+            'kind': 'import',
+            'sources': [repo.source_config(ref=ref)]
+        }
+        _yaml.dump(element, os.path.join(element_path, element_name))
+
+        context = Context()
+        context.load(config=user_config_file)
+        context.set_message_handler(message_handler)
+
+        project = Project(project_dir, context)
+        project.ensure_fully_loaded()
+
+        element = project.load_elements(['fetch.bst'])[0]
+        assert not element._source_cached()
+        source = list(element.sources())[0]
+
+        cas = context.get_cascache()
+        assert not cas.contains(source._get_source_name())
+
+        # Just check that we sensibly fetch and build the element
+        res = cli.run(project=project_dir, args=['build', 'fetch.bst'])
+        res.assert_success()
+
+        assert os.listdir(os.path.join(str(tmpdir), 'cache', 'sources', 'git')) != []
+
+        # Move source in local cas to repo
+        shutil.rmtree(os.path.join(str(tmpdir), 'sourceshare', 'repo', 'cas'))
+        shutil.move(
+            os.path.join(str(tmpdir), 'cache', 'cas'),
+            os.path.join(str(tmpdir), 'sourceshare', 'repo'))
+        shutil.rmtree(os.path.join(str(tmpdir), 'cache', 'sources'))
+
+        digest = share.cas.resolve_ref(source._get_source_name())
+        assert share.has_object(digest)
+
+        state = cli.get_element_state(project_dir, 'fetch.bst')
+        assert state == 'fetch needed'
+
+        # Now fetch the source and check
+        res = cli.run(project=project_dir, args=['source', 'fetch', 'fetch.bst'])
+        res.assert_success()
+        assert "Pulled source" in res.stderr
+
+        # check that we have the source in the cas now and it's not fetched
+        assert element._source_cached()
+        assert os.listdir(os.path.join(str(tmpdir), 'cache', 'sources', 'git')) == []
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_fetch_fallback(cli, tmpdir, datafiles):
+    project_dir = str(datafiles)
+
+    # use artifact cache for sources for now, they should work the same
+    with create_artifact_share(os.path.join(str(tmpdir), 'sourceshare')) as share:
+        # configure using this share
+        cache_dir = os.path.join(str(tmpdir), 'cache')
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+            },
+            'cachedir': cache_dir,
+        }
+        _yaml.dump(_yaml.node_sanitize(user_config), filename=user_config_file)
+        cli.configure(user_config)
+
+        repo = create_repo('git', str(tmpdir))
+        ref = repo.create(os.path.join(project_dir, 'files'))
+        element_path = os.path.join(project_dir, 'elements')
+        element_name = 'fetch.bst'
+        element = {
+            'kind': 'import',
+            'sources': [repo.source_config(ref=ref)]
+        }
+        _yaml.dump(element, os.path.join(element_path, element_name))
+
+        context = Context()
+        context.load(config=user_config_file)
+        context.set_message_handler(message_handler)
+
+        project = Project(project_dir, context)
+        project.ensure_fully_loaded()
+
+        element = project.load_elements(['fetch.bst'])[0]
+        assert not element._source_cached()
+        source = list(element.sources())[0]
+
+        cas = context.get_cascache()
+        assert not cas.contains(source._get_source_name())
+        assert not os.path.exists(os.path.join(cache_dir, 'sources'))
+
+        # Now check if it falls back to the source fetch method.
+        res = cli.run(project=project_dir, args=['source', 'fetch', 'fetch.bst'])
+        res.assert_success()
+        brief_key = source._get_brief_display_key()
+        assert ("Remote ({}) does not have source {} cached"
+                .format(share.repo, brief_key)) in res.stderr
+        assert ("SUCCESS Fetching from {}"
+                .format(repo.source_config(ref=ref)['url'])) in res.stderr
+
+        # Check that the source in both in the source dir and the local CAS
+        assert element._source_cached()
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_pull_fail(cli, tmpdir, datafiles):
+    project_dir = str(datafiles)
+    cache_dir = os.path.join(str(tmpdir), 'cache')
+
+    with create_artifact_share(os.path.join(str(tmpdir), 'sourceshare')) as share:
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+            },
+            'cachedir': cache_dir,
+        }
+        _yaml.dump(_yaml.node_sanitize(user_config), filename=user_config_file)
+        cli.configure(user_config)
+
+        repo = create_repo('git', str(tmpdir))
+        ref = repo.create(os.path.join(project_dir, 'files'))
+        element_path = os.path.join(project_dir, 'elements')
+        element_name = 'push.bst'
+        element = {
+            'kind': 'import',
+            'sources': [repo.source_config(ref=ref)]
+        }
+        _yaml.dump(element, os.path.join(element_path, element_name))
+
+        # get the source object
+        context = Context()
+        context.load(config=user_config_file)
+        context.set_message_handler(message_handler)
+        project = Project(project_dir, context)
+        project.ensure_fully_loaded()
+
+        element = project.load_elements(['push.bst'])[0]
+        assert not element._source_cached()
+        source = list(element.sources())[0]
+
+        # remove files and check that it doesn't build
+        shutil.rmtree(repo.repo)
+
+        # Should fail in stream, with a plugin tasks causing the error
+        res = cli.run(project=project_dir, args=['build', 'push.bst'])
+        res.assert_main_error(ErrorDomain.STREAM, None)
+        res.assert_task_error(ErrorDomain.PLUGIN, None)
+        assert "Remote ({}) does not have source {} cached".format(
+            share.repo, source._get_brief_display_key()) in res.stderr
diff --git a/tests/sourcecache/project/plugins/elements/always_fail.py b/tests/sourcecache/project/plugins/elements/always_fail.py
new file mode 100644
index 000000000..99ef0d7de
--- /dev/null
+++ b/tests/sourcecache/project/plugins/elements/always_fail.py
@@ -0,0 +1,32 @@
+#
+#  Copyright (C) 2019 Bloomberg Finance L.P.
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Authors:
+#        Raoul Hidalgo Charman <raoul.hidalgocharman@codethink.co.uk>
+#
+
+from buildstream.element import ElementError
+from buildstream.buildelement import BuildElement
+
+
+class AlwaysFail(BuildElement):
+
+    def assemble(self, sandbox):
+        raise ElementError("Always fails")
+
+
+def setup():
+    return AlwaysFail
diff --git a/tests/sourcecache/project/plugins/elements/always_fail.yaml b/tests/sourcecache/project/plugins/elements/always_fail.yaml
new file mode 100644
index 000000000..9934ab2f9
--- /dev/null
+++ b/tests/sourcecache/project/plugins/elements/always_fail.yaml
@@ -0,0 +1,22 @@
+# always-fail build element does not provide any default
+# build commands
+config:
+
+  # Commands for configuring the software
+  #
+  configure-commands: []
+
+  # Commands for building the software
+  #
+  build-commands: []
+
+  # Commands for installing the software into a
+  # destination folder
+  #
+  install-commands: []
+
+  # Commands for stripping installed binaries
+  #
+  strip-commands:
+  - |
+    %{strip-binaries}
diff --git a/tests/sourcecache/project/project.conf b/tests/sourcecache/project/project.conf
index 854e38693..728f3faa1 100644
--- a/tests/sourcecache/project/project.conf
+++ b/tests/sourcecache/project/project.conf
@@ -2,3 +2,10 @@
 name: test
 
 element-path: elements
+
+plugins:
+
+- origin: local
+  path: plugins/elements
+  elements:
+    always_fail: 0
\ No newline at end of file
diff --git a/tests/sourcecache/push.py b/tests/sourcecache/push.py
new file mode 100644
index 000000000..f692136bb
--- /dev/null
+++ b/tests/sourcecache/push.py
@@ -0,0 +1,222 @@
+#
+#  Copyright (C) 2019 Bloomberg Finance LP
+#
+#  This program is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 2 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
+#
+#  Authors:
+#        Raoul Hidalgo Charman <raoul.hidalgocharman@codethink.co.uk>
+#
+# Pylint doesn't play well with fixtures and dependency injection from pytest
+# pylint: disable=redefined-outer-name
+import os
+import shutil
+import pytest
+
+from buildstream._context import Context
+from buildstream._exceptions import ErrorDomain
+from buildstream._project import Project
+from buildstream import _yaml
+from buildstream.plugintestutils import cli  # pylint: disable=unused-import
+
+from tests.testutils import create_artifact_share, create_repo
+
+DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "project")
+
+
+def message_handler(message, context):
+    pass
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_source_push(cli, tmpdir, datafiles):
+    cache_dir = os.path.join(str(tmpdir), 'cache')
+    project_dir = str(datafiles)
+
+    with create_artifact_share(os.path.join(str(tmpdir), 'sourceshare')) as share:
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+                'push': True,
+            },
+            'cachedir': cache_dir,
+        }
+        _yaml.dump(_yaml.node_sanitize(user_config), filename=user_config_file)
+        cli.configure(user_config)
+
+        repo = create_repo('git', str(tmpdir))
+        ref = repo.create(os.path.join(project_dir, 'files'))
+        element_path = os.path.join(project_dir, 'elements')
+        element_name = 'push.bst'
+        element = {
+            'kind': 'import',
+            'sources': [repo.source_config(ref=ref)]
+        }
+        _yaml.dump(element, os.path.join(element_path, element_name))
+
+        # get the source object
+        context = Context()
+        context.load(config=user_config_file)
+        context.set_message_handler(message_handler)
+        project = Project(project_dir, context)
+        project.ensure_fully_loaded()
+
+        element = project.load_elements(['push.bst'])[0]
+        assert not element._source_cached()
+        source = list(element.sources())[0]
+
+        # check we don't have it in the current cache
+        cas = context.get_cascache()
+        assert not cas.contains(source._get_source_name())
+
+        # build the element, this should fetch and then push the source to the
+        # remote
+        res = cli.run(project=project_dir, args=['build', 'push.bst'])
+        res.assert_success()
+        assert "Pushed source" in res.stderr
+
+        # check that we've got the remote locally now
+        sourcecache = context.sourcecache
+        assert sourcecache.contains(source)
+
+        # check that's the remote CAS now has it
+        digest = share.cas.resolve_ref(source._get_source_name())
+        assert share.has_object(digest)
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_push_pull(cli, datafiles, tmpdir):
+    project_dir = str(datafiles)
+    cache_dir = os.path.join(str(tmpdir), 'cache')
+
+    with create_artifact_share(os.path.join(str(tmpdir), 'sourceshare')) as share:
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+                'push': True,
+            },
+            'cachedir': cache_dir,
+        }
+        _yaml.dump(_yaml.node_sanitize(user_config), filename=user_config_file)
+        cli.configure(user_config)
+
+        # create repo to pull from
+        repo = create_repo('git', str(tmpdir))
+        ref = repo.create(os.path.join(project_dir, 'files'))
+        element_path = os.path.join(project_dir, 'elements')
+        element_name = 'push.bst'
+        element = {
+            'kind': 'import',
+            'sources': [repo.source_config(ref=ref)]
+        }
+        _yaml.dump(element, os.path.join(element_path, element_name))
+
+        res = cli.run(project=project_dir, args=['build', 'push.bst'])
+        res.assert_success()
+
+        # remove local cache dir, and repo files and check it all works
+        shutil.rmtree(cache_dir)
+        os.makedirs(cache_dir)
+        shutil.rmtree(repo.repo)
+
+        # check it's pulls from the share
+        res = cli.run(project=project_dir, args=['build', 'push.bst'])
+        res.assert_success()
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_push_fail(cli, tmpdir, datafiles):
+    project_dir = str(datafiles)
+    cache_dir = os.path.join(str(tmpdir), 'cache')
+
+    # set up config with remote that we'll take down
+    with create_artifact_share(os.path.join(str(tmpdir), 'sourceshare')) as share:
+        remote = share.repo
+        user_config_file = str(tmpdir.join('buildstream.conf'))
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+                'push': True,
+            },
+            'cachedir': cache_dir,
+        }
+        _yaml.dump(_yaml.node_sanitize(user_config), filename=user_config_file)
+        cli.configure(user_config)
+
+    # create repo to pull from
+    repo = create_repo('git', str(tmpdir))
+    ref = repo.create(os.path.join(project_dir, 'files'))
+    element_path = os.path.join(project_dir, 'elements')
+    element_name = 'push.bst'
+    element = {
+        'kind': 'import',
+        'sources': [repo.source_config(ref=ref)]
+    }
+    _yaml.dump(element, os.path.join(element_path, element_name))
+
+    # build and check that it fails to set up the remote
+    res = cli.run(project=project_dir, args=['build', 'push.bst'])
+    res.assert_success()
+    assert ("Failed to initialize remote {}: Connect Failed"
+            .format(remote)) in res.stderr
+    assert "Pushing" not in res.stderr
+    assert "Pushed" not in res.stderr
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_source_push_build_fail(cli, tmpdir, datafiles):
+    project_dir = str(datafiles)
+    cache_dir = os.path.join(str(tmpdir), 'cache')
+
+    with create_artifact_share(os.path.join(str(tmpdir), 'share')) as share:
+        user_config = {
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+                'push': True,
+            },
+            'cachedir': cache_dir,
+        }
+        cli.configure(user_config)
+
+        repo = create_repo('git', str(tmpdir))
+        ref = repo.create(os.path.join(project_dir, 'files'))
+        element_path = os.path.join(project_dir, 'elements')
+
+        element_name = 'always-fail.bst'
+        element = {
+            'kind': 'always_fail',
+            'sources': [repo.source_config(ref=ref)]
+        }
+        _yaml.dump(element, os.path.join(element_path, element_name))
+
+        res = cli.run(project=project_dir, args=['build', 'always-fail.bst'])
+        res.assert_main_error(ErrorDomain.STREAM, None)
+        res.assert_task_error(ErrorDomain.ELEMENT, None)
+
+        # Sources are not pushed as the build queue is before the source push
+        # queue.
+        assert "Pushed source " not in res.stderr
diff --git a/tests/sourcecache/workspace.py b/tests/sourcecache/workspace.py
index 48ff3bf58..8bde39ee6 100644
--- a/tests/sourcecache/workspace.py
+++ b/tests/sourcecache/workspace.py
@@ -28,6 +28,7 @@ import pytest
 
 from buildstream.plugintestutils.runcli import cli  # pylint: disable=unused-import
 
+from tests.testutils.artifactshare import create_artifact_share
 from tests.testutils.element_generators import create_element_size
 
 
@@ -62,3 +63,42 @@ def test_workspace_source_fetch(tmpdir, datafiles, cli):
     assert 'Fetching from' in res.stderr
 
     assert os.listdir(workspace) != []
+
+
+@pytest.mark.datafiles(DATA_DIR)
+def test_workspace_open_no_source_push(tmpdir, datafiles, cli):
+    project_dir = os.path.join(str(tmpdir), 'project')
+    element_path = 'elements'
+    cache_dir = os.path.join(str(tmpdir), 'cache')
+    share_dir = os.path.join(str(tmpdir), 'share')
+    workspace = os.path.join(cli.directory, 'workspace')
+
+    with create_artifact_share(share_dir) as share:
+        cli.configure({
+            'cachedir': cache_dir,
+            'scheduler': {
+                'pushers': 1
+            },
+            'source-caches': {
+                'url': share.repo,
+                'push': True,
+            },
+        })
+
+        # Fetch as in previous test and check it pushes the source
+        create_element_size('target.bst', project_dir, element_path, [], 10000)
+        res = cli.run(project=project_dir, args=['build', 'target.bst'])
+        res.assert_success()
+        assert 'Fetching from' in res.stderr
+        assert 'Pushed source' in res.stderr
+
+        # clear the cas and open a workspace
+        shutil.rmtree(os.path.join(cache_dir, 'cas'))
+        res = cli.run(project=project_dir,
+                      args=['workspace', 'open', 'target.bst', '--directory', workspace])
+        res.assert_success()
+
+        # Check that this time it does not push the sources
+        res = cli.run(project=project_dir, args=['build', 'target.bst'])
+        res.assert_success()
+        assert "Pushed source" not in res.stderr