Merge lp:~frankban/juju-deployer/unit-errors into lp:juju-deployer

[Francesco Banconi]

Proceed with bundle deployments on unit errors.

Proceed with bundle deployments on unit errors.

This branch introduces a new watcher implementation
which allows for 1) watching the deployment process for
a subset of all the environment units, e.g. those
belonging to the bundle services and 2) react to unit
errors in different ways (e.g., a flag can be used to
make the importer proceed even if there are errors).

The current default behavior is maintained. The guiserver
code path instead configures the importer to complete the
deployment even if units are in an error state.

[Brad Crittenden (bac)]

Might be nice to mention in the help for '--ignore-errors' the relationship to retry_count.

Otherwise this branch looks really good. I'm -0 on changes like 'options = self.options' in existing code due to the noise for no real benefit, but "meh" as you say. :)

review: Approve (code)

[lp:~frankban/juju-deployer/unit-errors]

97. By Francesco Banconi on 2014-02-19

Changes as per review.

[Francesco Banconi (frankban)]

> Might be nice to mention in the help for '--ignore-errors' the relationship to
> retry_count.

Done.

> Otherwise this branch looks really good. I'm -0 on changes like 'options =
> self.options' in existing code due to the noise for no real benefit, but "meh"
> as you say. :)

Yeah, those kind of changes a far more readable e.g. in rietveld, where word-based diff helps reducing the noise.
Thanks for the review Brad.

[Kapil Thangavelu (hazmat)]

the watchers.log_on/warn_on/error_on seem extraneous given the feedback, and also at the wrong layer (part of the deployment via feedback api vs part of the environment interaction api).

[lp:~frankban/juju-deployer/unit-errors]

98. By Francesco Banconi on 2014-02-20

Merge trunk and resolve conflicts.

99. By Francesco Banconi on 2014-02-20

Fix tests.

=== modified file 'deployer/action/importer.py'
--- deployer/action/importer.py	2014-01-23 14:14:28 +0000
+++ deployer/action/importer.py	2014-02-20 12:33:52 +0000
@@ -2,6 +2,7 @@
 import time
 
 from .base import BaseAction
+from ..env import watchers
 from ..errors import UnitErrors
 from ..utils import ErrorExit
 
@@ -158,27 +159,32 @@
             return True
         return False
 
-    def wait_for_units(self, ignore_error=False):
+    def wait_for_units(self, ignore_errors=False):
         timeout = self.options.timeout - (time.time() - self.start_time)
         if timeout < 0:
             self.log.error("Reached deployment timeout.. exiting")
             raise ErrorExit()
-        try:
-            self.env.wait_for_units(
-                int(timeout), watch=self.options.watch, no_exit=ignore_error)
-        except UnitErrors:
-            if not ignore_error:
-                raise
+        # Set up the callback to be called in case of unit errors: if
+        # ignore_errors is True errors are just logged, otherwise we exit the
+        # program.
+        if ignore_errors:
+            on_errors = watchers.log_on_errors(self.env)
+        else:
+            on_errors = watchers.exit_on_errors(self.env)
+        self.env.wait_for_units(
+            int(timeout), watch=self.options.watch,
+            services=self.deployment.get_service_names(), on_errors=on_errors)
 
     def run(self):
+        options = self.options
         self.start_time = time.time()
-        if self.options.bootstrap:
+        if options.bootstrap:
             self.env.bootstrap()
         self.env.connect()
 
         # Get charms
         self.get_charms()
-        if self.options.branch_only:
+        if options.branch_only:
             return
 
         self.deploy_services()
@@ -189,20 +195,21 @@
         time.sleep(5.1)
         self.add_units()
 
+        ignore_errors = bool(options.retry_count) or options.ignore_errors
         self.log.debug("Waiting for units before adding relations")
-        self.wait_for_units()
+        self.wait_for_units(ignore_errors=ignore_errors)
 
         rels_created = self.add_relations()
 
         # Wait for the units to be up before waiting for rel stability.
         if rels_created:
             self.log.debug(
-                "Waiting for relation convergence %ds", self.options.rel_wait)
-            time.sleep(self.options.rel_wait)
-            self.wait_for_units(self.options.retry_count)
+                "Waiting for relation convergence %ds", options.rel_wait)
+            time.sleep(options.rel_wait)
+            self.wait_for_units(ignore_errors=ignore_errors)
 
-        if self.options.retry_count:
+        if options.retry_count:
             self.log.info("Looking for errors to auto-retry")
             self.env.resolve_errors(
-                self.options.retry_count,
-                self.options.timeout - time.time() - self.start_time)
+                options.retry_count,
+                options.timeout - time.time() - self.start_time)

=== modified file 'deployer/cli.py'
--- deployer/cli.py	2014-01-23 15:02:57 +0000
+++ deployer/cli.py	2014-02-20 12:33:52 +0000
@@ -102,6 +102,10 @@
         help=("Resolve CLI and unit errors via number of retries (default: 0)."
               " Either standalone or in a deployment"))
     parser.add_argument(
+        '--ignore-errors', action='store_true', dest='ignore_errors',
+        help='Proceed with the bundle deployment ignoring units errors. '
+             'Unit errors are also automatically ignored if --retry != 0')
+    parser.add_argument(
         "--diff", action="store_true", default=False,
         help=("Generate a delta between a configured deployment and a running"
               " environment."))

=== modified file 'deployer/deployment.py'
--- deployer/deployment.py	2013-11-20 05:08:15 +0000
+++ deployer/deployment.py	2014-02-20 12:33:52 +0000
@@ -47,6 +47,10 @@
         services.sort(self._placement_sort)
         return services
 
+    def get_service_names(self):
+        """Return a sequence of service names for this deployment."""
+        return self.data.get('services', {}).keys()
+
     @staticmethod
     def _placement_sort(svc_a, svc_b):
         if svc_a.unit_placement:

=== modified file 'deployer/env/go.py'
--- deployer/env/go.py	2013-11-27 21:05:17 +0000
+++ deployer/env/go.py	2014-02-20 12:33:52 +0000
@@ -2,14 +2,19 @@
 import socket
 import time
 
-from .base import BaseEnvironment
-from ..utils import ErrorExit
-
 from jujuclient import (
+    EnvError,
     Environment as EnvironmentClient,
     UnitErrors,
-    EnvError,
-    WatchWrapper)
+)
+
+from .base import BaseEnvironment
+from ..utils import ErrorExit
+from .watchers import (
+    raise_on_errors,
+    WaitForMachineTermination,
+    WaitForUnits,
+)
 
 
 class GoEnvironment(BaseEnvironment):
@@ -45,8 +50,8 @@
         while True:
             try:
                 self.client = EnvironmentClient(self.api_endpoint)
-            except socket.error, e:
-                if e.errno != errno.ETIMEDOUT:
+            except socket.error as err:
+                if err.errno != errno.ETIMEDOUT:
                     raise
                 continue
             else:
@@ -60,11 +65,10 @@
     def get_constraints(self, svc_name):
         try:
             return self.client.get_constraints(svc_name)
-        except EnvError, e:
-            if 'constraints do not apply to subordinate services' in str(e):
+        except EnvError as err:
+            if 'constraints do not apply to subordinate services' in str(err):
                 return {}
-            else:
-                raise e
+            raise
 
     def get_cli_status(self):
         status = super(GoEnvironment, self).get_cli_status()
@@ -94,7 +98,7 @@
             self.resolve_errors()
 
             # Wait for units
-            self.wait_for_units(timeout, "removed", watch=watch)
+            self.wait_for_units(timeout, goal_state='removed', watch=watch)
 
         # The only value to not terminating is keeping the data on the
         # machines around.
@@ -182,8 +186,8 @@
                 try:
                     self.client.resolved(e_uid, retry=bool(retry_count))
                     self.log.debug("  Resolving error on %s", e_uid)
-                except EnvError, e:
-                    if 'already resolved' in e:
+                except EnvError as err:
+                    if 'already resolved' in err:
                         continue
 
             if not error_units:
@@ -200,12 +204,13 @@
             count += 1
             try:
                 self.wait_for_units(
-                    timeout=int(w_timeout), watch=True, no_exit=True)
-            except UnitErrors, e:
+                    timeout=int(w_timeout), watch=True,
+                    on_errors=raise_on_errors(UnitErrors))
+            except UnitErrors as err:
                 if retry_count == count:
                     self.log.info(
                         " Retry count %d exhausted, but units in error (%s)",
-                        retry_count, " ".join(u['Name'] for u in e.errors))
+                        retry_count, " ".join(u['Name'] for u in err.errors))
                     return
             else:
                 return
@@ -216,24 +221,29 @@
     def status(self):
         return self.client.get_stat()
 
+    def log_errors(self, errors):
+        """Log the given unit errors.
+
+        This can be used in the WaitForUnits error handling machinery, e.g.
+        see deployer.watchers.log_on_errors.
+        """
+        messages = [
+            'unit: {Name}: machine: {MachineId} agent-state: {Status} '
+            'details: {StatusInfo}'.format(**error) for error in errors
+        ]
+        self.log.error(
+            'The following units had errors:\n   {}'.format(
+                '   \n'.join(messages)))
+
     def wait_for_units(
-            self, timeout, goal_state="started", watch=False, no_exit=False):
-        """Wait for units to reach a given condition.
-        """
-        callback = watch and self._delta_event_log or None
-        try:
-            self.client.wait_for_units(timeout, goal_state, callback=callback)
-        except UnitErrors, e:
-            error_units = [
-                "unit: %s: machine: %s agent-state: %s details: %s" % (
-                    u['Name'], u['MachineId'], u['Status'], u['StatusInfo']
-                )
-                for u in e.errors]
-            if no_exit:
-                raise
-            self.log.error("The following units had errors:\n   %s" % (
-                "   \n".join(error_units)))
-            raise ErrorExit()
+            self, timeout, goal_state="started", watch=False, services=None,
+            on_errors=None):
+        """Wait for units to reach a given condition."""
+        callback = self._delta_event_log if watch else None
+        watcher = self.client.get_watch(timeout)
+        WaitForUnits(
+            watcher, goal_state=goal_state,
+            services=services, on_errors=on_errors).run(callback)
 
     def _delta_event_log(self, et, ct, d):
         # event type, change type, data
@@ -260,25 +270,3 @@
             eps[0]['Relation']['Name'],
             eps[1]['ServiceName'],
             eps[1]['Relation']['Name'])
-
-
-class WaitForMachineTermination(WatchWrapper):
-
-    def __init__(self, watch, machines):
-        super(WaitForMachineTermination, self).__init__(watch)
-        self.machines = set(machines)
-        self.known = set()
-
-    def process(self, entity_type, change, data):
-        if entity_type != 'machine':
-            return
-        if change == 'remove' and data['Id'] in self.machines:
-            self.machines.remove(data['Id'])
-        else:
-            self.known.add(data['Id'])
-
-    def complete(self):
-        for m in self.machines:
-            if m in self.known:
-                return False
-        return True

=== modified file 'deployer/env/mem.py'
--- deployer/env/mem.py	2013-11-29 22:21:54 +0000
+++ deployer/env/mem.py	2014-02-20 12:33:52 +0000
@@ -168,6 +168,5 @@
         """ Return all services status """
         return {'services': self._services}
 
-    def wait_for_units(
-            self, timeout, goal_state="started", watch=False, no_exit=False):
+    def wait_for_units(self, *args, **kwargs):
         pass

=== modified file 'deployer/env/py.py'
--- deployer/env/py.py	2013-07-30 23:39:51 +0000
+++ deployer/env/py.py	2014-02-20 12:33:52 +0000
@@ -77,9 +77,26 @@
     def status(self):
         return self.get_cli_status()
 
+    def log_errors(self, errors):
+        """Log the given unit errors.
+
+        This can be used in the WaitForUnits error handling machinery, e.g.
+        see deployer.watchers.log_on_errors.
+        """
+        messages = [
+            'unit: {unit[name]}: machine: {unit[machine]} '
+            'agent-state: {unit[agent-state]}'.format(unit=error)
+            for error in errors
+        ]
+        self.log.error(
+            'The following units had errors:\n  {}'.format(
+                '   \n'.join(messages)))
+
     def wait_for_units(
-            self, timeout, goal_state="started", watch=False, no_exit=False):
-
+            self, timeout, goal_state="started", watch=False, services=None,
+            on_errors=None):
+        # Note that the services keyword argument is ignored in this pyJuju
+        # implementation: we wait for all the units in the environment.
         max_time = time.time() + timeout
         while max_time > time.time():
             status = self.status()
@@ -111,15 +128,4 @@
             if not pending and not errors:
                 break
             if errors:
-                if no_exit:
-                    raise UnitErrors(errors)
-                else:
-                    error_units = [
-                        "unit: %s: machine: %s agent-state: %s" % (
-                            u["name"], u["machine"], u["agent-state"]
-                            )
-                        for u in errors]
-                    self.log.error(
-                        "The following units had errors:\n  %s" % (
-                            "   \n".join(error_units)))
-                    raise ErrorExit()
+                on_errors(errors)

=== added file 'deployer/env/watchers.py'
--- deployer/env/watchers.py	1970-01-01 00:00:00 +0000
+++ deployer/env/watchers.py	2014-02-20 12:33:52 +0000
@@ -0,0 +1,110 @@
+"""A collection of juju-core environment watchers."""
+
+from jujuclient import WatchWrapper
+
+from ..utils import ErrorExit
+
+
+class WaitForMachineTermination(WatchWrapper):
+    """Wait until the given machines are terminated."""
+
+    def __init__(self, watch, machines):
+        super(WaitForMachineTermination, self).__init__(watch)
+        self.machines = set(machines)
+        self.known = set()
+
+    def process(self, entity_type, change, data):
+        if entity_type != 'machine':
+            return
+        if change == 'remove' and data['Id'] in self.machines:
+            self.machines.remove(data['Id'])
+        else:
+            self.known.add(data['Id'])
+
+    def complete(self):
+        for m in self.machines:
+            if m in self.known:
+                return False
+        return True
+
+
+class WaitForUnits(WatchWrapper):
+    """Wait for units of the environment to reach a particular goal state.
+
+    If services are provided, only consider the units belonging to the given
+    services.
+    If the on_errors callable is provided, call the given function each time a
+    change set is processed and a new unit is found in an error state. The
+    callable is called passing a list of units' data corresponding to the units
+    in an error state.
+    """
+    def __init__(
+            self, watch, goal_state='started', services=None, on_errors=None):
+        super(WaitForUnits, self).__init__(watch)
+        self.goal_state = goal_state
+        self.services = services
+        self.on_errors = on_errors
+        # The units dict maps unit names to units data.
+        self.units = {}
+        # The units_in_error list contains the names of the units in error.
+        self.units_in_error = []
+
+    def process(self, entity, action, data):
+        if entity != 'unit':
+            return
+        if (self.services is None) or (data['Service'] in self.services):
+            unit_name = data['Name']
+            if action == 'remove' and unit_name in self.units:
+                del self.units[unit_name]
+            else:
+                self.units[unit_name] = data
+
+    def complete(self):
+        ready = True
+        new_errors = []
+        goal_state = self.goal_state
+        on_errors = self.on_errors
+        units_in_error = self.units_in_error
+        for unit_name, data in self.units.items():
+            status = data['Status']
+            if status == 'error':
+                if unit_name not in units_in_error:
+                    units_in_error.append(unit_name)
+                    new_errors.append(data)
+            elif status != goal_state:
+                ready = False
+        if new_errors and goal_state != 'removed' and callable(on_errors):
+            on_errors(new_errors)
+        return ready
+
+
+def log_on_errors(env):
+    """Return a function receiving errors and logging them.
+
+    The resulting function is suitable to be used as the on_errors callback
+    for WaitForUnits (see above).
+    """
+    return env.log_errors
+
+
+def exit_on_errors(env):
+    """Return a function receiving errors, logging them and exiting the app.
+
+    The resulting function is suitable to be used as the on_errors callback
+    for WaitForUnits (see above).
+    """
+    def callback(errors):
+        log_on_errors(env)(errors)
+        raise ErrorExit()
+    return callback
+
+
+def raise_on_errors(exception):
+    """Return a function receiving errors and raising the given exception.
+
+    The resulting function is suitable to be used as the on_errors callback
+    for WaitForUnits (see above).
+    """
+    def callback(errors):
+        raise exception(errors)
+    return callback

=== modified file 'deployer/guiserver.py'
--- deployer/guiserver.py	2014-02-12 21:00:47 +0000
+++ deployer/guiserver.py	2014-02-20 12:33:52 +0000
@@ -12,6 +12,7 @@
 import os
 
 from deployer.action.importer import Importer
+from deployer.cli import setup_parser
 from deployer.deployment import Deployment
 from deployer.env.gui import GUIEnvironment
 from deployer.utils import (
@@ -26,6 +27,13 @@
 JUJU_HOME = '/var/lib/juju-gui/juju-home'
 
 
+def get_default_guiserver_options():
+    """Return the default importer options used by the GUI server."""
+    # Options used by the juju-deployer.  The defaults work for us, except for
+    # the ignore_errors flag.
+    return setup_parser().parse_args(['--ignore-errors'])
+
+
 class GUIDeployment(Deployment):
     """Handle bundle deployments requested by the GUI server."""
 

=== modified file 'deployer/tests/test_deployment.py'
--- deployer/tests/test_deployment.py	2014-01-23 12:55:11 +0000
+++ deployer/tests/test_deployment.py	2014-02-20 12:33:52 +0000
@@ -148,3 +148,11 @@
         self.assertEqual(
             [('keystone', 'mysql'), ('glance', 'mysql'),
              ('nova-compute', 'mysql')], list(d.get_relations()))
+
+    def test_getting_service_names(self):
+        # It is possible to retrieve the service names.
+        deployment = self.get_named_deployment("stack-placement.yaml", "stack")
+        service_names = deployment.get_service_names()
+        expected_service_names = [
+            'ceph', 'mysql', 'nova-compute', 'quantum', 'semper', 'verity']
+        self.assertEqual(set(expected_service_names), set(service_names))

=== modified file 'deployer/tests/test_guiserver.py'
--- deployer/tests/test_guiserver.py	2014-02-20 01:20:23 +0000
+++ deployer/tests/test_guiserver.py	2014-02-20 12:33:52 +0000
@@ -10,14 +10,57 @@
 import mock
 import yaml
 
-from deployer import (
-    cli,
-    guiserver,
-)
+from deployer import guiserver
 from deployer.feedback import Feedback
 from deployer.tests.base import TEST_OFFLINE
 
 
+class TestGetDefaultGuiserverOptions(unittest.TestCase):
+
+    def setUp(self):
+        self.options = guiserver.get_default_guiserver_options()
+
+    def test_option_keys(self):
+        # All the required options are returned.
+        # When adding/modifying options, ensure the defaults are sane for us.
+        expected_keys = set([
+            'bootstrap', 'branch_only', 'configs', 'debug', 'deploy_delay',
+            'deployment', 'description', 'destroy_services', 'diff',
+            'find_service', 'ignore_errors', 'juju_env', 'list_deploys',
+            'no_local_mods', 'overrides', 'rel_wait', 'retry_count', 'series',
+            'terminate_machines', 'timeout', 'update_charms', 'verbose',
+            'watch'
+        ])
+        self.assertEqual(expected_keys, set(self.options.__dict__.keys()))
+
+    def test_option_values(self):
+        # The options values are suitable to be used by the GUI server.
+        # When adding/modifying options, ensure the defaults are sane for us.
+        options = self.options
+        self.assertFalse(options.bootstrap)
+        self.assertFalse(options.branch_only)
+        self.assertIsNone(options.configs)
+        self.assertFalse(options.debug)
+        self.assertEqual(0, options.deploy_delay)
+        self.assertIsNone(options.deployment)
+        self.assertFalse(options.destroy_services)
+        self.assertFalse(options.diff)
+        self.assertIsNone(options.find_service)
+        self.assertTrue(options.ignore_errors)
+        self.assertEqual(os.getenv("JUJU_ENV"), options.juju_env)
+        self.assertFalse(options.list_deploys)
+        self.assertTrue(options.no_local_mods)
+        self.assertIsNone(options.overrides)
+        self.assertEqual(60, options.rel_wait)
+        self.assertEqual(0, options.retry_count)
+        self.assertIsNone(options.series)
+        self.assertFalse(options.terminate_machines)
+        self.assertEqual(2700, options.timeout)
+        self.assertFalse(options.update_charms)
+        self.assertFalse(options.verbose)
+        self.assertFalse(options.watch)
+
+
 class TestDeploymentError(unittest.TestCase):
 
     def test_error(self):
@@ -152,7 +195,7 @@
 class TestImportBundle(DeployerFunctionsTestMixin, unittest.TestCase):
 
     # The options attribute simulates the options passed to the Importer.
-    options = cli.setup_parser().parse_args([])
+    options = guiserver.get_default_guiserver_options()
 
     @contextmanager
     def patch_juju_home(self):
@@ -274,30 +317,3 @@
             'Invalid config charm cs:precise/mysql-28 bad=wolf',
         ])
         self.assertEqual(expected_errors, set(cm.exception.errors))
-
-    def test_options(self, mock_environment):
-        # Ensure the default options are sane for us.
-        expected = argparse.Namespace(
-            bootstrap=False,
-            branch_only=False,
-            configs=None,
-            debug=False,
-            deploy_delay=0,
-            deployment=None,
-            description=False,
-            destroy_services=False,
-            diff=False,
-            find_service=None,
-            juju_env=os.environ.get("JUJU_ENV"),
-            list_deploys=False,
-            no_local_mods=True,
-            overrides=None,
-            rel_wait=60,
-            retry_count=0,
-            series=None,
-            terminate_machines=False,
-            timeout=2700,
-            update_charms=False,
-            verbose=False,
-            watch=False)
-        self.assertEqual(expected, self.options)

=== modified file 'deployer/tests/test_pyenv.py'
--- deployer/tests/test_pyenv.py	2013-07-22 15:29:31 +0000
+++ deployer/tests/test_pyenv.py	2014-02-20 12:33:52 +0000
@@ -1,14 +1,14 @@
-
 import StringIO
 
+from jujuclient import UnitErrors
+
+from .base import Base
+from deployer.env import watchers
 from deployer.env.py import PyEnvironment
 from deployer.errors import UnitErrors
 from deployer.utils import setup_logging, ErrorExit
 
 
-from .base import Base
-
-
 class FakePyEnvironment(PyEnvironment):
 
     def __init__(self, name, status):
@@ -38,16 +38,15 @@
                },
               },
              })
-        try:
-            env.wait_for_units(timeout=240, no_exit=True)
-        except UnitErrors, e:
-            self.assertEqual(1, len(e.errors))
-            unit = e.errors[0]
-            self.assertEqual("wordpress/0", unit["name"])
-            self.assertEqual("install-error", unit["agent-state"])
-            self.assertEqual(1, unit["machine"])
-        else:
-            self.fail("Did not get expected exception")
+        with self.assertRaises(UnitErrors) as cm:
+            env.wait_for_units(
+                timeout=240, on_errors=watchers.raise_on_errors(UnitErrors))
+        errors = cm.exception.errors
+        self.assertEqual(1, len(errors))
+        unit = errors[0]
+        self.assertEqual("wordpress/0", unit["name"])
+        self.assertEqual("install-error", unit["agent-state"])
+        self.assertEqual(1, unit["machine"])
 
     def test_wait_for_units_error_exit(self):
         env = FakePyEnvironment(
@@ -62,16 +61,13 @@
                },
               },
              })
-        try:
-            env.wait_for_units(timeout=240, no_exit=False)
-        except ErrorExit:
-            self.assertIn('The following units had errors:',
-                          self.output.getvalue())
-            self.assertIn('unit: wordpress/0: machine: 1 '
-                          'agent-state: install-error',
-                          self.output.getvalue())
-        else:
-            self.fail("Did not get expected exception")
+        with self.assertRaises(ErrorExit):
+            env.wait_for_units(
+                timeout=240, on_errors=watchers.exit_on_errors(env))
+        output = self.output.getvalue()
+        self.assertIn('The following units had errors:', output)
+        self.assertIn(
+            'unit: wordpress/0: machine: 1 agent-state: install-error', output)
 
     def test_wait_for_units_sub_error_no_exit(self):
         env = FakePyEnvironment(
@@ -91,16 +87,15 @@
                },
               },
              })
-        try:
-            env.wait_for_units(timeout=240, no_exit=True)
-        except UnitErrors, e:
-            self.assertEqual(1, len(e.errors))
-            unit = e.errors[0]
-            self.assertEqual("nrpe/0", unit["name"])
-            self.assertEqual("install-error", unit["agent-state"])
-            self.assertEqual(1, unit["machine"])
-        else:
-            self.fail("Did not get expected exception")
+        with self.assertRaises(UnitErrors) as cm:
+            env.wait_for_units(
+                timeout=240, on_errors=watchers.raise_on_errors(UnitErrors))
+        errors = cm.exception.errors
+        self.assertEqual(1, len(errors))
+        unit = errors[0]
+        self.assertEqual("nrpe/0", unit["name"])
+        self.assertEqual("install-error", unit["agent-state"])
+        self.assertEqual(1, unit["machine"])
 
     def test_wait_for_units_no_error_no_exit(self):
         env = FakePyEnvironment(
@@ -121,9 +116,10 @@
               },
              })
         try:
-            env.wait_for_units(timeout=240, no_exit=True)
-        except UnitErrors, e:
-            self.fail("Unexpected exception: %s" % e)
+            env.wait_for_units(
+                timeout=240, on_errors=watchers.raise_on_errors(UnitErrors))
+        except UnitErrors as err:
+            self.fail("Unexpected exception: %s" % err)
 
     def test_wait_for_units_relation_error_no_exit(self):
         env = FakePyEnvironment(
@@ -141,16 +137,15 @@
                },
               },
              })
-        try:
-            env.wait_for_units(timeout=240, no_exit=True)
-        except UnitErrors, e:
-            self.assertEqual(1, len(e.errors))
-            unit = e.errors[0]
-            self.assertEqual("wordpress/0", unit["name"])
-            self.assertEqual("relation-error: nrpe", unit["agent-state"])
-            self.assertEqual(1, unit["machine"])
-        else:
-            self.fail("Did not get expected exception")
+        with self.assertRaises(UnitErrors) as cm:
+            env.wait_for_units(
+                timeout=240, on_errors=watchers.raise_on_errors(UnitErrors))
+        errors = cm.exception.errors
+        self.assertEqual(1, len(errors))
+        unit = errors[0]
+        self.assertEqual("wordpress/0", unit["name"])
+        self.assertEqual("relation-error: nrpe", unit["agent-state"])
+        self.assertEqual(1, unit["machine"])
 
     def test_wait_for_units_subordinate_no_unit_no_error_no_exit(self):
         env = FakePyEnvironment(
@@ -163,6 +158,7 @@
               },
              })
         try:
-            env.wait_for_units(timeout=240, no_exit=True)
-        except UnitErrors, e:
-            self.fail("Unexpected exception: %s" % e)
+            env.wait_for_units(
+                timeout=240, on_errors=watchers.raise_on_errors(UnitErrors))
+        except UnitErrors as err:
+            self.fail("Unexpected exception: %s" % err)

=== added file 'deployer/tests/test_watchers.py'
--- deployer/tests/test_watchers.py	1970-01-01 00:00:00 +0000
+++ deployer/tests/test_watchers.py	2014-02-20 12:33:52 +0000
@@ -0,0 +1,146 @@
+"""Tests juju-core environment watchers."""
+
+import unittest
+
+import mock
+
+from deployer.env import watchers
+from deployer.utils import ErrorExit
+
+
+class TestWaitForUnits(unittest.TestCase):
+
+    def make_watch(self, *changesets):
+        """Create and return a mock watcher returning the given change sets."""
+        watch = mock.MagicMock()
+        watch.__iter__.return_value = changesets
+        return watch
+
+    def make_change(self, unit_name, status):
+        """Create and return a juju-core mega-watcher unit change."""
+        service = unit_name.split('/')[0]
+        return {'Name': unit_name, 'Service': service, 'Status': status}
+
+    def test_success(self):
+        # It is possible to watch for units to be started.
+        watch = self.make_watch(
+            [('unit', 'change', self.make_change('django/0', 'pending')),
+             ('unit', 'change', self.make_change('haproxy/1', 'pending'))],
+            [('unit', 'change', self.make_change('django/0', 'started')),
+             ('unit', 'change', self.make_change('haproxy/1', 'started'))],
+        )
+        watcher = watchers.WaitForUnits(watch)
+        callback = mock.Mock()
+        watcher.run(callback)
+        # The callback has been called once for each change in the change sets,
+        # excluding the initial state.
+        self.assertEqual(2, callback.call_count)
+        # The watcher has been properly stopped.
+        watch.stop.assert_called_once_with()
+
+    def test_errors_handling(self):
+        # Errors can be handled providing an on_errors callable.
+        watch = self.make_watch(
+            [('unit', 'change', self.make_change('django/0', 'pending')),
+             ('unit', 'change', self.make_change('django/42', 'pending')),
+             ('unit', 'change', self.make_change('haproxy/1', 'pending'))],
+            [('unit', 'change', self.make_change('django/0', 'pending')),
+             ('unit', 'change', self.make_change('django/42', 'error')),
+             ('unit', 'change', self.make_change('haproxy/1', 'error'))],
+            [('unit', 'change', self.make_change('django/0', 'error')),
+             ('unit', 'change', self.make_change('django/42', 'error')),
+             ('unit', 'change', self.make_change('haproxy/1', 'error'))],
+        )
+        on_errors = mock.Mock()
+        watcher = watchers.WaitForUnits(watch, on_errors=on_errors)
+        watcher.run()
+        # The watcher has been properly stopped.
+        watch.stop.assert_called_once_with()
+        # The errors handler has been called once for each changeset containing
+        # errors.
+        self.assertEqual(2, on_errors.call_count)
+        on_errors.assert_has_calls([
+            mock.call([
+                {'Status': 'error', 'Name': 'django/42', 'Service': 'django'},
+                {'Status': 'error', 'Name': 'haproxy/1', 'Service': 'haproxy'}
+            ]),
+            mock.call([
+                {'Status': 'error', 'Name': 'django/0', 'Service': 'django'}]),
+        ])
+
+    def test_specific_services(self):
+        # It is possible to only watch units belonging to specific services.
+        watch = self.make_watch(
+            [('unit', 'change', self.make_change('django/0', 'pending')),
+             ('unit', 'change', self.make_change('django/42', 'pending')),
+             ('unit', 'change', self.make_change('haproxy/1', 'pending')),
+             ('unit', 'change', self.make_change('haproxy/47', 'pending'))],
+            [('unit', 'change', self.make_change('django/0', 'pending')),
+             ('unit', 'change', self.make_change('django/42', 'started')),
+             ('unit', 'change', self.make_change('haproxy/1', 'error')),
+             ('unit', 'change', self.make_change('haproxy/47', 'pending'))],
+            [('unit', 'change', self.make_change('django/0', 'error')),
+             ('unit', 'change', self.make_change('django/42', 'started')),
+             ('unit', 'change', self.make_change('haproxy/1', 'error')),
+             ('unit', 'change', self.make_change('haproxy/47', 'pending'))],
+        )
+        on_errors = mock.Mock()
+        watcher = watchers.WaitForUnits(
+            watch, services=['django'], on_errors=on_errors)
+        watcher.run()
+        # The watcher has been properly stopped, even if haproxy/47 is pending.
+        watch.stop.assert_called_once_with()
+        # The errors handler has been called for the django error, not for the
+        # haproxy one.
+        on_errors.assert_called_once_with(
+            [{'Status': 'error', 'Name': 'django/0', 'Service': 'django'}])
+
+    def test_goal_states(self):
+        # It is possible to watch for the given goal state
+        watch = self.make_watch(
+            [('unit', 'change', self.make_change('django/0', 'pending')),
+             ('unit', 'change', self.make_change('haproxy/1', 'pending'))],
+        )
+        watcher = watchers.WaitForUnits(watch, goal_state='pending')
+        callback = mock.Mock()
+        watcher.run(callback)
+        # Since all the units are already pending, the watcher has been
+        # properly stopped.
+        watch.stop.assert_called_once_with()
+
+
+class TestLogOnErrors(unittest.TestCase):
+
+    def setUp(self):
+        # Set up a mock environment.
+        self.env = mock.Mock()
+
+    def test_returned_callable(self):
+        # The returned function uses the env to log errors.
+        callback = watchers.log_on_errors(self.env)
+        self.assertEqual(self.env.log_errors, callback)
+
+
+class TestExitOnErrors(unittest.TestCase):
+
+    def setUp(self):
+        # Set up a mock environment.
+        self.env = mock.Mock()
+
+    def test_returned_callable(self):
+        # The returned function uses the env to log errors and then exits the
+        # application.
+        callback = watchers.exit_on_errors(self.env)
+        with self.assertRaises(ErrorExit):
+            callback('bad wolf')
+        self.env.log_errors.assert_called_once_with('bad wolf')
+
+
+class TestRaiseOnErrors(unittest.TestCase):
+
+    def test_returned_callable(self):
+        # The returned function raises the given exception passing the errors.
+        callback = watchers.raise_on_errors(ValueError)
+        with self.assertRaises(ValueError) as cm:
+            callback('bad wolf')
+        self.assertEqual('bad wolf', bytes(cm.exception))