Adds basic test for convert command.

bonobo/commands/convert.py

@@ -6,8 +6,8 @@ from bonobo.util.resolvers import _resolve_transformations, _resolve_options
 
 class ConvertCommand(BaseCommand):
     def add_arguments(self, parser):
-        parser.add_argument('input-filename', help='Input filename.')
-        parser.add_argument('output-filename', help='Output filename.')
+        parser.add_argument('input_filename', help='Input filename.')
+        parser.add_argument('output_filename', help='Output filename.')
         parser.add_argument(
             '--' + READER,
             '-r',

bonobo/execution/contexts/base.py

@@ -1,3 +1,4 @@
+import logging
 import sys
 from contextlib import contextmanager
 from logging import WARNING, ERROR
@@ -38,6 +39,10 @@ class LoopingExecutionContext(Wrapper):
     def stopped(self):
         return self._stopped
 
+    @property
+    def defunct(self):
+        return self._defunct
+
     @property
     def alive(self):
         return self._started and not self._stopped
@@ -45,6 +50,8 @@ class LoopingExecutionContext(Wrapper):
     @property
     def status(self):
         """One character status for this node. """
+        if self._defunct:
+            return '!'
         if not self.started:
             return ' '
         if not self.stopped:
@@ -65,7 +72,7 @@ class LoopingExecutionContext(Wrapper):
         else:
             self.services = None
 
-        self._started, self._stopped = False, False
+        self._started, self._stopped, self._defunct = False, False, False
         self._stack = None
 
     def __enter__(self):
@@ -81,15 +88,17 @@ class LoopingExecutionContext(Wrapper):
 
         self._started = True
 
-        self._stack = ContextCurrifier(self.wrapped, *self._get_initial_context())
-        if isconfigurabletype(self.wrapped):
-            # Not normal to have a partially configured object here, so let's warn the user instead of having get into
-            # the hard trouble of understanding that by himself.
-            raise TypeError(
-                'The Configurable should be fully instanciated by now, unfortunately I got a PartiallyConfigured object...'
-            )
-
-        self._stack.setup(self)
+        try:
+            self._stack = ContextCurrifier(self.wrapped, *self._get_initial_context())
+            if isconfigurabletype(self.wrapped):
+                # Not normal to have a partially configured object here, so let's warn the user instead of having get into
+                # the hard trouble of understanding that by himself.
+                raise TypeError(
+                    'The Configurable should be fully instanciated by now, unfortunately I got a PartiallyConfigured object...'
+                )
+            self._stack.setup(self)
+        except Exception:
+            return self.fatal(sys.exc_info())
 
     def loop(self):
         """Generic loop. A bit boring. """
@@ -113,14 +122,17 @@ class LoopingExecutionContext(Wrapper):
         finally:
             self._stopped = True
 
-    def handle_error(self, exctype, exc, tb):
-        mondrian.excepthook(
-            exctype, exc, tb, level=WARNING, context='{} in {}'.format(exctype.__name__, get_name(self)), logger=logger
-        )
-
     def _get_initial_context(self):
         if self.parent:
             return self.parent.services.args_for(self.wrapped)
         if self.services:
             return self.services.args_for(self.wrapped)
         return ()
+
+    def handle_error(self, exctype, exc, tb, *, level=logging.ERROR):
+        logging.getLogger(__name__).log(level, repr(self), exc_info=(exctype, exc, tb))
+
+    def fatal(self, exc_info):
+        self._defunct = True
+        self.input.shutdown()
+        self.handle_error(*exc_info, level=logging.CRITICAL)

bonobo/execution/contexts/node.py

@@ -1,3 +1,4 @@
+import logging
 import sys
 from queue import Empty
 from time import sleep
@@ -12,6 +13,7 @@ from bonobo.structs.tokens import Token
 from bonobo.util import get_name, iserrorbag, isloopbackbag, isbag, istuple
 from bonobo.util.compat import deprecated_alias
 from bonobo.util.statistics import WithStatistics
+from mondrian import term
 
 
 class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
@@ -39,10 +41,12 @@ class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
         return '<{}({}{}){}>'.format(type_name, self.status, name, self.get_statistics_as_string(prefix=' '))
 
     def get_flags_as_string(self):
+        if self._defunct:
+            return term.red('[defunct]')
         if self.killed:
-            return '[killed]'
+            return term.lightred('[killed]')
         if self.stopped:
-            return '[done]'
+            return term.lightblack('[done]')
         return ''
 
     def write(self, *messages):
@@ -92,13 +96,13 @@ class NodeExecutionContext(WithStatistics, LoopingExecutionContext):
         self.increment('in')
         return row
 
+    def should_loop(self):
+        return not any((self.defunct, self.killed))
+
     def loop(self):
-        while not self._killed:
+        while self.should_loop():
             try:
                 self.step()
-            except KeyboardInterrupt:
-                self.handle_error(*sys.exc_info())
-                break
             except InactiveReadableError:
                 break
             except Empty:

bonobo/execution/strategies/executor.py

@@ -27,7 +27,11 @@ class ExecutorStrategy(Strategy):
         futures = []
 
         with self.create_executor() as executor:
-            context.start(self.get_starter(executor, futures))
+            try:
+                context.start(self.get_starter(executor, futures))
+            except:
+                logging.getLogger(__name__
+                                  ).warning('KeyboardInterrupt received. Trying to terminate the nodes gracefully.')
 
             while context.alive:
                 try:
@@ -50,12 +54,17 @@ class ExecutorStrategy(Strategy):
                 try:
                     with node:
                         node.loop()
-                except BaseException as exc:
-                    logging.getLogger(__name__).info(
-                        'Got {} in {} runner.'.format(get_name(exc), node), exc_info=sys.exc_info()
+                except:
+                    logging.getLogger(__name__).critical(
+                        'Uncaught exception in node execution for {}.'.format(node), exc_info=True
                     )
+                    node.shutdown()
+                    node.stop()
 
-            futures.append(executor.submit(_runner))
+            try:
+                futures.append(executor.submit(_runner))
+            except:
+                logging.getLogger(__name__).critical('futures.append', exc_info=sys.exc_info())
 
         return starter
 

bonobo/nodes/io/base.py

@@ -1,4 +1,7 @@
+from fs.errors import ResourceNotFound
+
 from bonobo.config import Configurable, ContextProcessor, Option, Service
+from bonobo.errors import UnrecoverableError
 
 
 class FileHandler(Configurable):

bonobo/util/environ.py

@@ -152,3 +152,13 @@ def parse_args(mixed=None):
                 del os.environ[name]
             else:
                 os.environ[name] = value
+
+
+@contextmanager
+def change_working_directory(path):
+    old_dir = os.getcwd()
+    os.chdir(str(path))
+    try:
+        yield
+    finally:
+        os.chdir(old_dir)

bonobo/util/resolvers.py

@@ -72,6 +72,7 @@ def _resolve_transformations(transformations):
     :return: tuple(object)
     """
     registry = _ModulesRegistry()
+    transformations = transformations or []
     for t in transformations:
         try:
             mod, attr = t.split(':', 1)

bonobo/util/testing.py

@@ -1,6 +1,15 @@
-from contextlib import contextmanager
+import functools
+import io
+import os
+import runpy
+import sys
+from contextlib import contextmanager, redirect_stdout, redirect_stderr
+from unittest.mock import patch
 
-from bonobo import open_fs, Token
+import pytest
+
+from bonobo import open_fs, Token, __main__, get_examples_path
+from bonobo.commands import entrypoint
 from bonobo.execution.contexts.graph import GraphExecutionContext
 from bonobo.execution.contexts.node import NodeExecutionContext
 
@@ -64,3 +73,68 @@ class BufferingGraphExecutionContext(BufferingContext, GraphExecutionContext):
 
     def create_node_execution_context_for(self, node):
         return self.NodeExecutionContextType(node, parent=self, buffer=self.buffer)
+
+
+def runner(f):
+    @functools.wraps(f)
+    def wrapped_runner(*args, catch_errors=False):
+        with redirect_stdout(io.StringIO()) as stdout, redirect_stderr(io.StringIO()) as stderr:
+            try:
+                f(list(args))
+            except BaseException as exc:
+                if not catch_errors:
+                    raise
+                elif isinstance(catch_errors, BaseException) and not isinstance(exc, catch_errors):
+                    raise
+                return stdout.getvalue(), stderr.getvalue(), exc
+        return stdout.getvalue(), stderr.getvalue()
+
+    return wrapped_runner
+
+
+@runner
+def runner_entrypoint(args):
+    """ Run bonobo using the python command entrypoint directly (bonobo.commands.entrypoint). """
+    return entrypoint(args)
+
+
+@runner
+def runner_module(args):
+    """ Run bonobo using the bonobo.__main__ file, which is equivalent as doing "python -m bonobo ..."."""
+    with patch.object(sys, 'argv', ['bonobo', *args]):
+        return runpy.run_path(__main__.__file__, run_name='__main__')
+
+
+all_runners = pytest.mark.parametrize('runner', [runner_entrypoint, runner_module])
+all_environ_targets = pytest.mark.parametrize(
+    'target', [
+        (get_examples_path('environ.py'), ),
+        (
+            '-m',
+            'bonobo.examples.environ',
+        ),
+    ]
+)
+
+
+@all_runners
+@all_environ_targets
+class EnvironmentTestCase():
+    def run_quiet(self, runner, *args):
+        return runner('run', '--quiet', *args)
+
+    def run_environ(self, runner, *args, environ=None):
+        _environ = {'PATH': '/usr/bin'}
+        if environ:
+            _environ.update(environ)
+
+        with patch.dict('os.environ', _environ, clear=True):
+            out, err = self.run_quiet(runner, *args)
+            assert 'SECRET' not in os.environ
+            assert 'PASSWORD' not in os.environ
+            if 'PATH' in _environ:
+                assert 'PATH' in os.environ
+                assert os.environ['PATH'] == _environ['PATH']
+
+        assert err == ''
+        return dict(map(lambda line: line.split(' ', 1), filter(None, out.split('\n'))))