_stream.py: Add StreamError exception

Use Stream error for Stream errors.

3 files changed, 44 insertions, 33 deletions

diff --git a/buildstream/_exceptions.py b/buildstream/_exceptions.py
index d8687104f..bcea65a8d 100644
--- a/buildstream/_exceptions.py
+++ b/buildstream/_exceptions.py
@@ -88,6 +88,7 @@ class ErrorDomain(Enum):
     SOURCE = 10
     ELEMENT = 11
     APP = 12
+    STREAM = 13
 
 
 # BstError is an internal base exception class for BuildSream
@@ -252,10 +253,20 @@ class ArtifactError(BstError):
 
 # PipelineError
 #
-# Raised when a pipeline fails
+# Raised from pipeline operations
 #
 class PipelineError(BstError):
 
+    def __init__(self, message, *, detail=None, reason=None):
+        super().__init__(message, detail=detail, domain=ErrorDomain.PIPELINE, reason=reason)
+
+
+# StreamError
+#
+# Raised when a stream operation fails
+#
+class StreamError(BstError):
+
     def __init__(self, message=None, *, detail=None, reason=None, terminated=False):
 
         # The empty string should never appear to a user,
@@ -264,7 +275,7 @@ class PipelineError(BstError):
         if message is None:
             message = ""
 
-        super().__init__(message, detail=detail, domain=ErrorDomain.PIPELINE, reason=reason)
+        super().__init__(message, detail=detail, domain=ErrorDomain.STREAM, reason=reason)
 
         self.terminated = terminated
 
diff --git a/buildstream/_frontend/app.py b/buildstream/_frontend/app.py
index 84c33a385..feac0ab9c 100644
--- a/buildstream/_frontend/app.py
+++ b/buildstream/_frontend/app.py
@@ -37,7 +37,7 @@ from .. import Scope, Consistency
 # Import various buildstream internals
 from .._context import Context
 from .._project import Project
-from .._exceptions import BstError, PipelineError, LoadError, LoadErrorReason, AppError
+from .._exceptions import BstError, StreamError, LoadError, LoadErrorReason, AppError
 from .._message import Message, MessageType, unconditional_messages
 from .._stream import Stream
 from .._pipeline import Pipeline, PipelineSelection
@@ -316,7 +316,7 @@ class App():
                 elapsed = self.scheduler.elapsed_time()
 
                 if session_name:
-                    if isinstance(e, PipelineError) and e.terminated:  # pylint: disable=no-member
+                    if isinstance(e, StreamError) and e.terminated:  # pylint: disable=no-member
                         self._message(MessageType.WARN, session_name + ' Terminated', elapsed=elapsed)
                     else:
                         self._message(MessageType.FAIL, session_name, elapsed=elapsed)
@@ -484,10 +484,10 @@ class App():
             self.stream.fetch(self.scheduler, [target])
 
         if not no_checkout and target._get_consistency() != Consistency.CACHED:
-            raise PipelineError("Could not stage uncached source. " +
-                                "Use `--track` to track and " +
-                                "fetch the latest version of the " +
-                                "source.")
+            raise AppError("Could not stage uncached source. " +
+                           "Use `--track` to track and " +
+                           "fetch the latest version of the " +
+                           "source.")
 
         try:
             os.makedirs(directory, exist_ok=True)
diff --git a/buildstream/_stream.py b/buildstream/_stream.py
index d80b19f34..725c3836f 100644
--- a/buildstream/_stream.py
+++ b/buildstream/_stream.py
@@ -23,7 +23,7 @@ import shlex
 import tarfile
 from tempfile import TemporaryDirectory
 
-from ._exceptions import PipelineError, ImplError, BstError
+from ._exceptions import StreamError, ImplError, BstError
 from . import _site
 from . import utils
 from . import Scope, Consistency
@@ -67,9 +67,9 @@ class Stream():
 
         _, status = self._scheduler.run([track])
         if status == SchedStatus.ERROR:
-            raise PipelineError()
+            raise StreamError()
         elif status == SchedStatus.TERMINATED:
-            raise PipelineError(terminated=True)
+            raise StreamError(terminated=True)
 
     # fetch()
     #
@@ -108,9 +108,9 @@ class Stream():
 
         _, status = self._scheduler.run(queues)
         if status == SchedStatus.ERROR:
-            raise PipelineError()
+            raise StreamError()
         elif status == SchedStatus.TERMINATED:
-            raise PipelineError(terminated=True)
+            raise StreamError(terminated=True)
 
     # build()
     #
@@ -168,9 +168,9 @@ class Stream():
 
         _, status = self._scheduler.run(queues)
         if status == SchedStatus.ERROR:
-            raise PipelineError()
+            raise StreamError()
         elif status == SchedStatus.TERMINATED:
-            raise PipelineError(terminated=True)
+            raise StreamError(terminated=True)
 
     # checkout()
     #
@@ -190,14 +190,14 @@ class Stream():
         try:
             os.makedirs(directory, exist_ok=True)
         except OSError as e:
-            raise PipelineError("Failed to create checkout directory: {}".format(e)) from e
+            raise StreamError("Failed to create checkout directory: {}".format(e)) from e
 
         if not os.access(directory, os.W_OK):
-            raise PipelineError("Directory {} not writable".format(directory))
+            raise StreamError("Directory {} not writable".format(directory))
 
         if not force and os.listdir(directory):
-            raise PipelineError("Checkout directory is not empty: {}"
-                                .format(directory))
+            raise StreamError("Checkout directory is not empty: {}"
+                              .format(directory))
 
         # Stage deps into a temporary sandbox first
         try:
@@ -212,10 +212,10 @@ class Stream():
                         else:
                             utils.copy_files(sandbox_root, directory)
                     except OSError as e:
-                        raise PipelineError("Failed to checkout files: {}".format(e)) from e
+                        raise StreamError("Failed to checkout files: {}".format(e)) from e
         except BstError as e:
-            raise PipelineError("Error while staging dependencies into a sandbox: {}".format(e),
-                                reason=e.reason) from e
+            raise StreamError("Error while staging dependencies into a sandbox: {}".format(e),
+                              reason=e.reason) from e
 
     # Helper function for checkout()
     #
@@ -223,7 +223,7 @@ class Stream():
         try:
             removed = utils.safe_remove(directory)
         except OSError as e:
-            raise PipelineError("Failed to remove checkout directory: {}".format(e)) from e
+            raise StreamError("Failed to remove checkout directory: {}".format(e)) from e
 
         if removed:
             # Try a simple rename of the sandbox root; if that
@@ -247,7 +247,7 @@ class Stream():
     def pull(self, scheduler, elements):
 
         if not self._pipeline._artifacts.has_fetch_remotes():
-            raise PipelineError("Not artifact caches available for pulling artifacts")
+            raise StreamError("Not artifact caches available for pulling artifacts")
 
         plan = elements
         self._pipeline._assert_consistent(plan)
@@ -259,9 +259,9 @@ class Stream():
 
         _, status = self._scheduler.run(queues)
         if status == SchedStatus.ERROR:
-            raise PipelineError()
+            raise StreamError()
         elif status == SchedStatus.TERMINATED:
-            raise PipelineError(terminated=True)
+            raise StreamError(terminated=True)
 
     # push()
     #
@@ -274,7 +274,7 @@ class Stream():
     def push(self, scheduler, elements):
 
         if not self._pipeline._artifacts.has_push_remotes():
-            raise PipelineError("No artifact caches available for pushing artifacts")
+            raise StreamError("No artifact caches available for pushing artifacts")
 
         plan = elements
         self._pipeline._assert_consistent(plan)
@@ -286,9 +286,9 @@ class Stream():
 
         _, status = self._scheduler.run(queues)
         if status == SchedStatus.ERROR:
-            raise PipelineError()
+            raise StreamError()
         elif status == SchedStatus.TERMINATED:
-            raise PipelineError(terminated=True)
+            raise StreamError(terminated=True)
 
     # source_bundle()
     #
@@ -316,8 +316,8 @@ class Stream():
             open(tar_location, mode="x")
             os.remove(tar_location)
         except IOError as e:
-            raise PipelineError("Cannot write to {0}: {1}"
-                                .format(tar_location, e)) from e
+            raise StreamError("Cannot write to {0}: {1}"
+                              .format(tar_location, e)) from e
 
         plan = list(dependencies)
         self.fetch(self._scheduler, plan)
@@ -334,8 +334,8 @@ class Stream():
             try:
                 os.makedirs(source_directory)
             except OSError as e:
-                raise PipelineError("Failed to create directory: {}"
-                                    .format(e)) from e
+                raise StreamError("Failed to create directory: {}"
+                                  .format(e)) from e
 
             # Any elements that don't implement _write_script
             # should not be included in the later stages.