_stream.py: Ensure push does not fail if artifact not cachedjennis/push_unbuilt_artifact

A test for this has also been added to tests/frontend/push.py

1 files changed, 35 insertions, 6 deletions

diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 41a90f8ea..c54fee1a7 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -40,7 +40,7 @@ from ._scheduler import Scheduler, SchedStatus, TrackQueue, FetchQueue, \
 from ._pipeline import Pipeline, PipelineSelection
 from ._profile import Topics, PROFILER
 from ._state import State
-from .types import _KeyStrength
+from .types import _KeyStrength, _SchedulerErrorAction
 from . import utils, _yaml, _site
 from . import Scope, Consistency
 
@@ -470,11 +470,40 @@ class Stream():
             self._add_queue(PullQueue(self._scheduler))
             self._enqueue_plan(require_buildtrees)
 
-        self._scheduler.clear_queues()
-        push_queue = ArtifactPushQueue(self._scheduler)
-        self._add_queue(push_queue)
-        self._enqueue_plan(elements, queue=push_queue)
-        self._run()
+        # Before we try to push the artifacts, ensure they're cached
+        cached_elements = []
+        uncached_elements = []
+        self._message(MessageType.INFO, "Verifying that elements are cached")
+        for element in elements:
+            if element._cached():
+                cached_elements.append(element)
+            else:
+                msg = "{} is not cached".format(element.name)
+                if self._context.sched_error_action != _SchedulerErrorAction.CONTINUE:
+                    raise StreamError("Push failed: " + msg)
+                else:
+                    self._message(MessageType.WARN, msg)
+                    uncached_elements.append(element)
+
+        if cached_elements:
+            self._scheduler.clear_queues()
+            push_queue = ArtifactPushQueue(self._scheduler)
+            self._add_queue(push_queue)
+            self._enqueue_plan(cached_elements, queue=push_queue)
+            self._run()
+
+        # If the user has selected to continue on error, fail the command
+        # and print a summary of artifacts which could not be pushed
+        #
+        # NOTE: Usually we check the _SchedulerErrorAction when a *job* has failed.
+        #       However, we cannot create a PushQueue job unless we intentionally
+        #       ready an uncached element in the PushQueue.
+        if self._context.sched_error_action == _SchedulerErrorAction.CONTINUE and uncached_elements:
+            names = [element.name for element in uncached_elements]
+            fail_str = "Error while pushing. The following elements were not pushed as they are " \
+                "not yet cached:\n\n\t{}\n".format("\n\t".join(names))
+
+            raise StreamError(fail_str)
 
     # checkout()
     #