Merge lp:~fwereade/pyjuju/watch-related-units-callbacks into lp:pyjuju

[William Reade]

Lack of separate callbacks for membership changes and settings changes made it harder for me to follow the code, and especially the tests. Should now be more comprehensible. Both callbacks still need to be synchronous, but this doesn't represent a real change IMO; it's just something to remain aware of.

[William Reade (fwereade)]

rejecting due to abandonment of prereq

[lp:~fwereade/pyjuju/watch-related-units-callbacks]

485. By William Reade on 2011-12-12

merge parent

486. By William Reade on 2011-12-12

merge parent

487. By William Reade on 2011-12-15

merge parent

488. By William Reade on 2011-12-16

merge parent

489. By William Reade on 2012-01-11

merge parent

490. By William Reade on 2012-01-11

rename foo_callback to cb_foo

491. By William Reade on 2012-01-11

poke_zk changes, as requested (since these are just checking for callbacks firing, I think they're suitable)

492. By William Reade on 2012-01-13

merge parent

493. By William Reade on 2012-01-13

merge parent

494. By William Reade on 2012-01-13

merge parent

495. By William Reade on 2012-01-16

merge parent

496. By William Reade on 2012-01-16

merge parent

497. By William Reade on 2012-01-23

merge parent

498. By William Reade on 2012-01-27

merge parent

499. By William Reade on 2012-02-02

merge parent

500. By William Reade on 2012-02-08

merge parent

501. By William Reade on 2012-02-21

merge parent

Unmerged revisions

=== modified file 'juju/hooks/scheduler.py'
--- juju/hooks/scheduler.py	2011-12-05 02:11:54 +0000
+++ juju/hooks/scheduler.py	2011-12-05 02:11:55 +0000
@@ -137,31 +137,11 @@
         # occurs.
         self._run_queue.put(None)
 
-    def notify_change(self, old_units=(), new_units=(), modified=()):
-        """Receive changes regarding related units and schedule hook execution.
-        """
-        log.debug("relation change old:%s, new:%s, modified:%s",
-                  old_units, new_units, modified)
-
+    def members_changed(self, old_units, new_units):
+        log.debug("members changed: old=%s, new=%s", old_units, new_units)
+        scheduled = 0
         self._clock_sequence += 1
 
-        # UnitRelationState.watch_related_units will EITHER call back with
-        # old_units and new_units, OR with modified; never both.
-        # This should in fact probably be two distinct callbacks...
-        if old_units or new_units:
-            assert not modified, "got mixed settings and membership changes"
-            scheduled = self._notify_membership_change(old_units, new_units)
-        else:
-            assert modified, "got no changes"
-            scheduled = self._notify_settings_change(modified)
-
-        if scheduled:
-            self._run_queue.put(self._clock_sequence)
-        self._save_state()
-
-    def _notify_membership_change(self, old_units, new_units):
-        scheduled = 0
-
         if self._context_members is None:
             self._context_members = list(old_units)
 
@@ -184,16 +164,22 @@
             scheduled += self._queue_change(
                 unit_name, REMOVED, self._clock_sequence)
 
-        return scheduled
+        if scheduled:
+            self._run_queue.put(self._clock_sequence)
+        self._save_state()
 
-    def _notify_settings_change(self, modified):
+    def settings_changed(self, unit_versions):
+        log.debug("settings changed: %s", unit_versions)
         scheduled = 0
-        for (unit_name, version) in modified:
+        self._clock_sequence += 1
+        for (unit_name, version) in unit_versions:
             if version > self._latest_members.get(unit_name, 0):
                 self._latest_members[unit_name] = version
                 scheduled += self._queue_change(
                     unit_name, MODIFIED, self._clock_sequence)
-        return scheduled
+        if scheduled:
+            self._run_queue.put(self._clock_sequence)
+        self._save_state()
 
     def get_hook_context(self, change):
         """

=== modified file 'juju/hooks/tests/test_scheduler.py'
--- juju/hooks/tests/test_scheduler.py	2011-12-05 02:11:54 +0000
+++ juju/hooks/tests/test_scheduler.py	2011-12-05 02:11:55 +0000
@@ -56,14 +56,14 @@
         results in a modify event.
         """
         self.write_single_unit_state()
-        self.scheduler.notify_change(old_units=["u-1"], new_units=[])
-        self.scheduler.notify_change(old_units=[], new_units=["u-1"])
+        self.scheduler.members_changed(["u-1"], [])
+        self.scheduler.members_changed([], ["u-1"])
         self.scheduler.run()
         self.assertEqual(len(self.executions), 1)
         self.assertEqual(self.executions[0][1].change_type, "modified")
 
-        output = ("relation change old:['u-1'], new:[], modified:()",
-                  "relation change old:[], new:['u-1'], modified:()",
+        output = ("members changed: old=['u-1'], new=[]",
+                  "members changed: old=[], new=['u-1']",
                   "start",
                   "executing hook for u-1:modified\n")
         self.assertEqual(self.log_stream.getvalue(), "\n".join(output))
@@ -73,33 +73,33 @@
         An extra validation of the previous test.
         """
         self.write_single_unit_state()
-        self.scheduler.notify_change(modified=[("u-1", 1)])
-        self.scheduler.notify_change(old_units=["u-1"], new_units=[])
-        self.scheduler.notify_change(old_units=[], new_units=["u-1"])
+        self.scheduler.settings_changed([("u-1", 1)])
+        self.scheduler.members_changed(["u-1"], [])
+        self.scheduler.members_changed([], ["u-1"])
         self.scheduler.run()
         self.assertEqual(len(self.executions), 1)
         self.assertEqual(self.executions[0][1].change_type, "modified")
 
     def test_reduce_add_modify(self):
         """An add and modify event for a node are coalesced to an add."""
-        self.scheduler.notify_change(old_units=[], new_units=["u-1"])
-        self.scheduler.notify_change(modified=[("u-1", 1)])
+        self.scheduler.members_changed([], ["u-1"])
+        self.scheduler.settings_changed([("u-1", 1)])
         self.scheduler.run()
         self.assertEqual(len(self.executions), 1)
         self.assertEqual(self.executions[0][1].change_type, "joined")
 
     def test_reduce_add_remove(self):
         """an add followed by a removal results in a noop."""
-        self.scheduler.notify_change(old_units=[], new_units=["u-1"])
-        self.scheduler.notify_change(old_units=["u-1"], new_units=[])
+        self.scheduler.members_changed([], ["u-1"])
+        self.scheduler.members_changed(["u-1"], [])
         self.scheduler.run()
         self.assertEqual(len(self.executions), 0)
 
     def test_reduce_modify_remove(self):
         """Modifying and then removing a node, results in just the removal."""
         self.write_single_unit_state()
-        self.scheduler.notify_change(modified=[("u-1", 1)])
-        self.scheduler.notify_change(old_units=["u-1"], new_units=[])
+        self.scheduler.settings_changed([("u-1", 1)])
+        self.scheduler.members_changed(["u-1"], [])
         self.scheduler.run()
         self.assertEqual(len(self.executions), 1)
         self.assertEqual(self.executions[0][1].change_type, "departed")
@@ -108,15 +108,15 @@
         """Multiple modifies get coalesced to a single modify."""
         # simulate normal startup, the first notify will always be the existing
         # membership set.
-        self.scheduler.notify_change(old_units=[], new_units=["u-1"])
+        self.scheduler.members_changed([], ["u-1"])
         self.scheduler.run()
         self.scheduler.stop()
         self.assertEqual(len(self.executions), 1)
 
         # Now continue the modify/modify reduction.
-        self.scheduler.notify_change(modified=[("u-1", 1)])
-        self.scheduler.notify_change(modified=[("u-1", 2)])
-        self.scheduler.notify_change(modified=[("u-1", 3)])
+        self.scheduler.settings_changed([("u-1", 1)])
+        self.scheduler.settings_changed([("u-1", 2)])
+        self.scheduler.settings_changed([("u-1", 3)])
         self.scheduler.run()
 
         self.assertEqual(len(self.executions), 2)
@@ -143,11 +143,9 @@
         always see the membership of a relation as it was at the
         time of their associated change.
         """
-        self.scheduler.notify_change(
-            old_units=[], new_units=["u-1", "u-2"])
-        self.scheduler.notify_change(
-            old_units=["u-1", "u-2"], new_units=["u-2", "u-3"])
-        self.scheduler.notify_change(modified=[("u-2", 1)])
+        self.scheduler.members_changed([], ["u-1", "u-2"])
+        self.scheduler.members_changed(["u-1", "u-2"], ["u-2", "u-3"])
+        self.scheduler.settings_changed([("u-2", 1)])
 
         self.scheduler.run()
         self.scheduler.stop()
@@ -161,9 +159,8 @@
         change_members = yield self.executions[0][0].get_members()
         self.assertEqual(change_members, ["u-2"])
 
-        self.scheduler.notify_change(modified=[("u-2", 2)])
-        self.scheduler.notify_change(
-            old_units=["u-2", "u-3"], new_units=["u-2"])
+        self.scheduler.settings_changed([("u-2", 2)])
+        self.scheduler.members_changed(["u-2", "u-3"], ["u-2"])
         self.scheduler.run()
 
         self.assertEqual(len(self.executions), 4)
@@ -187,10 +184,8 @@
                 "clock_queue": [],
                 "clock_sequence": 1}))
 
-        self.scheduler.notify_change(
-            old_units=["u-1", "u-2"],
-            new_units=["u-2", "u-3", "u-4"])
-        self.scheduler.notify_change(modified=[("u-2", 1)])
+        self.scheduler.members_changed(["u-1", "u-2"], ["u-2", "u-3", "u-4"])
+        self.scheduler.settings_changed([("u-2", 1)])
 
         self.scheduler.run()
         self.scheduler.stop()
@@ -273,10 +268,8 @@
                 "clock_queue": [],
                 "clock_sequence": 7}))
 
-        self.scheduler.notify_change(
-            old_units=["u-1", "u-2"],
-            new_units=["u-2", "u-3"])
-        self.scheduler.notify_change(modified=[("u-2", 3)])
+        self.scheduler.members_changed(["u-1", "u-2"], ["u-2", "u-3"])
+        self.scheduler.settings_changed([("u-2", 3)])
 
         # Add a stop instruction to the queue, which should *not* be saved.
         self.scheduler.stop()
@@ -358,7 +351,7 @@
         version for that unit will be ignored.
         """
         self.write_single_unit_state()
-        self.scheduler.notify_change(modified=(("u-1", 0),))
+        self.scheduler.settings_changed([("u-1", 0),])
         self.scheduler.run()
         self.assertEquals(len(self.executions), 0)
 
@@ -367,8 +360,8 @@
         When a unit is added, we assume its settings version to be 0, and
         therefore modified events with version 0 will be ignored.
         """
-        self.scheduler.notify_change(old_units=[], new_units=["u-1"])
-        self.scheduler.notify_change(modified=(("u-1", 0),))
+        self.scheduler.members_changed([], ["u-1"])
+        self.scheduler.settings_changed([("u-1", 0),])
         self.scheduler.run()
         self.assertEquals(len(self.executions), 1)
         self.assertEqual(self.executions[0][1].change_type, "joined")
@@ -387,12 +380,11 @@
             "clock_queue": [],
             "clock_sequence": 4}))
 
-        self.scheduler.notify_change(
-            old_units=["u-2"], new_units=["u-3", "u-4"])
+        self.scheduler.members_changed(["u-2"], ["u-3", "u-4"])
         self.scheduler.run()
 
         output = (
-            "relation change old:['u-2'], new:['u-3', 'u-4'], modified:()",
+            "members changed: old=['u-2'], new=['u-3', 'u-4']",
             "old does not match last recorded units: ['u-1', 'u-2']",
             "start",
             "executing hook for u-3:joined",

=== modified file 'juju/state/relation.py'
--- juju/state/relation.py	2011-12-05 02:11:54 +0000
+++ juju/state/relation.py	2011-12-05 02:11:55 +0000
@@ -408,7 +408,7 @@
         returnValue(endpoint_container)
 
     @inlineCallbacks
-    def watch_related_units(self, callback):
+    def watch_related_units(self, members_callback, settings_callback):
         """Register a callback to be invoked when related units change.
 
         @param: callback a function that gets invoked when related
@@ -443,7 +443,8 @@
             watcher_factory = PeerUnitWatcher
 
         watcher = watcher_factory(
-            self._client, self, endpoint_container, callback)
+            self._client, self, endpoint_container,
+            members_callback, settings_callback)
         returnValue(watcher)
 
 
@@ -474,12 +475,14 @@
                  client,
                  watcher_unit,
                  unit_container_path,
-                 callback):
+                 members_callback,
+                 settings_callback):
         super(RelationUnitWatcherBase, self).__init__(client)
         self._units = []
         self._watcher_unit = watcher_unit
         self._container_path = unit_container_path
-        self._callback = callback
+        self._members_callback = members_callback
+        self._settings_callback = settings_callback
         self._stopped = None
         self._unit_name_map = None
         self._log = logging.getLogger("unit.relation.watch")
@@ -606,9 +609,7 @@
         # Invoke callback
         callback_d.addCallback(
             lambda (old_units, new_units): maybeDeferred(
-                self._callback,
-                old_units=sorted(old_units),
-                new_units=sorted(new_units)))
+                self._members_callback, sorted(old_units), sorted(new_units)))
 
         # Attach initial notifiers and change handlers to new nodes
         if settings_watches:
@@ -631,7 +632,7 @@
             return
         ((unit_name,),) = yield self._resolve_unit_names([unit_id])
         node_info = (unit_name, stat["version"])
-        yield self._callback(modified=(node_info,))
+        yield self._settings_callback((node_info,))
 
     def _watch_settings_for_units(self, added):
         """Setup watches on new unit relation setting nodes.

=== modified file 'juju/state/tests/test_relation.py'
--- juju/state/tests/test_relation.py	2011-12-05 02:11:54 +0000
+++ juju/state/tests/test_relation.py	2011-12-05 02:11:55 +0000
@@ -24,6 +24,18 @@
 from juju.state.tests.common import StateTestBase
 
 
+def _1_arg(f):
+    def g(arg1):
+        return f(arg1)
+    return g
+
+
+def _2_args(f):
+    def g(arg1, arg2):
+        return f(arg1, arg2)
+    return g
+
+
 class RelationTestBase(StateTestBase):
 
     @inlineCallbacks
@@ -716,23 +728,60 @@
                               self.service_state2)))
 
 
+class WatchChecker(object):
+
+    def __init__(self, test, block_cbs=False):
+        self.test = test
+        self.results = []
+        self.sentinels = [Deferred() for i in range(10)]
+        self.blockers = [Deferred() if block_cbs else True for i in range(10)]
+
+    def _members_changed(self, old, new):
+        self.results.append((old, new))
+        return self._synchronize()
+
+    def _settings_changed(self, modified):
+        (change,) = modified
+        self.results.append(change)
+        return self._synchronize()
+
+    @inlineCallbacks
+    def _synchronize(self):
+        index = len(self.results) - 1
+        self.sentinels[index].callback(True)
+        yield self.blockers[index]
+
+    def watch(self, unit_relation):
+        return unit_relation.watch_related_units(
+            self._members_changed, self._settings_changed)
+
+    def assert_cb_count(self, count):
+        # *started* callbacks, including those whose return is blocked
+        self.test.assertEquals(len(self.results), count)
+
+    def wait_for_cb(self, index):
+        # wait for *call*, not return
+        return self.sentinels[index]
+
+    def unblock_cb(self, index):
+        self.blockers[index].callback(True)
+
+    @inlineCallbacks
+    def assert_members_cb(self, index, old, new):
+        yield self.wait_for_cb(index)
+        for actual, expected in zip(self.results[index], (old, new)):
+            for actual_name, expected_unit in zip(actual, expected):
+                self.test.assertEquals(actual_name, expected_unit.unit_name)
+
+    @inlineCallbacks
+    def assert_settings_cb(self, index, unit, version):
+        yield self.wait_for_cb(index)
+        change = self.results[index]
+        self.test.assertEquals(change, (unit.unit_name, version))
+
+
 class UnitRelationStateTest(RelationTestBase):
 
-    def verify_unit_watch_result(
-            self, result, old_units=None, new_units=None, modified=None):
-        old_actual, new_actual, modified_actual = result
-        for value, expected in zip(result, [old_units, new_units, modified]):
-            if expected is None:
-                self.assertFalse(value)
-                continue
-            for value_unit, expected_unit in zip(value, expected):
-                if isinstance(expected_unit, tuple):
-                    self.assertEqual(
-                        value_unit[0], expected_unit[0].unit_name)
-                    self.assertEqual(value_unit[1], expected_unit[1])
-                else:
-                    self.assertEqual(value_unit, expected_unit.unit_name)
-
     @inlineCallbacks
     def test_properties(self):
         states = yield self.add_relation_service_unit("webcache", "varnish")
@@ -822,36 +871,29 @@
 
         yield self.add_opposite_service_unit(wordpress_states)
 
-        results = []
-        wait_callback = [Deferred() for i in range(5)]
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        def invoked(*args, **kw):
-            # sleep to make sure that things haven't fired till the watch is
-            # in place.
-            time.sleep(0.1)
-            self.assertFalse(results)
-
+        checker = WatchChecker(self)
         mock_client = self.mocker.patch(self.client)
         mock_client.get_children_and_watch("/relations/%s/server" % (
             wordpress_states["relation"].internal_id))
 
+        def invoked(*args, **kw):
+            # sleep to make sure that things haven't fired till the watch is
+            # in place.
+            time.sleep(0.1)
+            checker.assert_cb_count(0)
+
         self.mocker.call(invoked)
         self.mocker.passthrough()
         self.mocker.replay()
 
-        watcher = yield wordpress_states["unit_relation"].watch_related_units(
-            watch_related)
+        watcher = yield checker.watch(wordpress_states["unit_relation"])
         yield watcher.start()
-        yield wait_callback[0]
+        yield checker.wait_for_cb(0)
 
     @inlineCallbacks
     def test_watch_start_new_service(self):
         """Invoking watcher.start returns a deferred that only fires
-        after watch on the containr is in place. In the case of a new
+        after watch on the container is in place. In the case of a new
         service this after an existance watch is established on the
         container.
         """
@@ -860,28 +902,23 @@
         wordpress_states = yield self.add_relation_service_unit_from_endpoints(
             wordpress_ep)
 
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-
-        def invoked(*args, **kw):
-            # sleep to make sure that things haven't fired till the watch is
-            # in place.
-            time.sleep(0.1)
-            self.assertFalse(results)
-
+        checker = WatchChecker(self)
         mock_client = self.mocker.patch(self.client)
         mock_client.exists_and_watch("/relations/%s/server" % (
             wordpress_states["relation"].internal_id))
 
+        def invoked(*args, **kw):
+            # sleep to make sure that things haven't fired till the watch is
+            # in place.
+            time.sleep(0.1)
+            checker.assert_cb_count(0)
+
         self.mocker.call(invoked)
         self.mocker.passthrough()
         self.mocker.replay()
 
-        watcher = yield wordpress_states["unit_relation"].watch_related_units(
-            watch_related)
-        yield watcher.start().addCallback(lambda x: results.append(True))
+        watcher = yield checker.watch(wordpress_states["unit_relation"])
+        yield watcher.start()
 
     @inlineCallbacks
     def test_watch_client_server_with_new_service(self):
@@ -896,16 +933,8 @@
         wordpress_states = yield self.add_relation_service_unit_from_endpoints(
             wordpress_ep, mysql_ep)
 
-        # setup watch callbacks, and start watching.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results) - 1].callback(True)
-
-        watcher = yield wordpress_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(wordpress_states["unit_relation"])
         yield watcher.start()
 
         # adding another unit of wordpress, does not cause any changes
@@ -915,13 +944,11 @@
 
         # give chance for accidental watch firing.
         yield self.sleep(0.1)
-
-        # assert no firing of callback
-        self.assertFalse(results)
+        checker.assert_cb_count(0)
 
         # add the server service and a unit of that
-        mysql_states = yield self.add_opposite_service_unit(
-            wordpress_states)
+        mysql_states = yield self.add_opposite_service_unit(wordpress_states)
+        mysql_unit = mysql_states["unit"]
 
         topology = yield self.get_topology()
         # assert the relation is established correctly
@@ -929,30 +956,19 @@
             wordpress_states["relation"].internal_id)
         self.assertEqual(len(services), 2)
 
-        # wait for callback
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[mysql_states["unit"]])
-
-        # wait for initial settings version
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-             results[1], modified=((mysql_states["unit"], 0),))
-
-        # modify the unit and get another callback
+        # wait for initial callbacks
+        yield checker.assert_members_cb(0, [], [mysql_unit])
+        yield checker.assert_settings_cb(1, mysql_unit, 0)
+
+        # modify unit, wait for callback
         yield mysql_states["unit_relation"].set_data(dict(hello="world"))
-        yield wait_callback[2]
-        self.verify_unit_watch_result(
-             results[2], modified=((mysql_states["unit"], 1),))
+        yield checker.assert_settings_cb(2, mysql_unit, 1)
 
-        # add another unit of mysql
+        # add another unit, wait for callback
         mysql_unit2 = yield mysql_states["service"].add_unit_state()
         yield mysql_states["service_relation"].add_unit_state(mysql_unit2)
-        yield wait_callback[3]
-        self.verify_unit_watch_result(
-            results[3],
-            old_units=[mysql_states["unit"]],
-            new_units=[mysql_states["unit"], mysql_unit2])
+        yield checker.assert_members_cb(
+            3, [mysql_unit], [mysql_unit, mysql_unit2])
 
     @inlineCallbacks
     def test_watch_client_server_with_existing_service(self):
@@ -969,34 +985,15 @@
             wordpress_ep, mysql_ep)
         mysql_states = yield self.add_opposite_service_unit(
             wordpress_states)
-
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        watcher = yield wordpress_states["unit_relation"].watch_related_units(
-            watch_related)
+        mysql_unit = mysql_states["unit"]
+
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(wordpress_states["unit_relation"])
         yield watcher.start()
-
-        # wait for initial callback
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[mysql_states["unit"]])
-
-        # and initial settings version notification
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((mysql_states["unit"], 0),))
-
-        # modify the unit and get another callback, verify the result.
+        yield checker.assert_members_cb(0, [], [mysql_unit])
+        yield checker.assert_settings_cb(1, mysql_unit, 0)
         yield mysql_states["unit_relation"].set_data(dict(hello="world"))
-        yield wait_callback[2]
-        self.verify_unit_watch_result(
-            results[2], modified=((mysql_states["unit"], 1),))
+        yield checker.assert_settings_cb(2, mysql_unit, 1)
 
         # directly delete the presence node to trigger a deletion notification
         self.client.delete("/relations/%s/server/%s" % (
@@ -1004,9 +1001,7 @@
             mysql_states["unit"].internal_id))
 
         # verify the deletion result.
-        yield wait_callback[3]
-        self.verify_unit_watch_result(
-            results[3], old_units=[mysql_states["unit"]], new_units=[])
+        yield checker.assert_members_cb(3, [mysql_unit], [])
 
     @inlineCallbacks
     def test_watch_server_client_with_new_service(self):
@@ -1020,32 +1015,18 @@
         mysql_states = yield self.add_relation_service_unit_from_endpoints(
             mysql_ep, wordpress_ep)
 
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        watcher = yield mysql_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(mysql_states["unit_relation"])
         yield watcher.start()
-
-        # assert no firing of callback
-        self.assertFalse(results)
+        yield self.sleep(0.1)
+        checker.assert_cb_count(0)
 
         # add the client service and a unit of that
         wordpress_states = yield self.add_opposite_service_unit(
             mysql_states)
-
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[wordpress_states["unit"]])
-
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((wordpress_states["unit"], 0),))
+        wordpress_unit = wordpress_states["unit"]
+        yield checker.assert_members_cb(0, [], [wordpress_unit])
+        yield checker.assert_settings_cb(1, wordpress_unit, 0)
 
     @inlineCallbacks
     def test_watch_peer(self):
@@ -1059,58 +1040,37 @@
         riak2_unit = yield riak_states["service"].add_unit_state()
         yield riak_states["service_relation"].add_unit_state(riak2_unit)
 
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(10)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        watcher = yield riak_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(riak_states["unit_relation"])
         yield watcher.start()
 
         # wait for initial callbacks
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[riak2_unit])
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((riak2_unit, 0),))
+        yield checker.assert_members_cb(0, [], [riak2_unit])
+        yield checker.assert_settings_cb(1, riak2_unit, 0)
 
         # verify modifying self does not cause a notification.
         yield riak_states["unit_relation"].set_data(dict(hello="world"))
         yield self.sleep(0.1)
-        self.assertEqual(len(results), 2)
+        checker.assert_cb_count(2)
 
         # add another unit
         riak3_unit = yield riak_states["service"].add_unit_state()
         riak3_relation = yield riak_states["service_relation"].add_unit_state(
             riak3_unit)
-
-        yield wait_callback[2]
-        self.verify_unit_watch_result(
-            results[2],
-            old_units=[riak2_unit], new_units=[riak2_unit, riak3_unit])
-        yield wait_callback[3]
-        self.verify_unit_watch_result(
-            results[3], modified=((riak3_unit, 0),))
-
-        # add remove one (no api atm, so directly to trigger notification)
+        yield checker.assert_members_cb(
+            2, [riak2_unit], [riak2_unit, riak3_unit])
+        yield checker.assert_settings_cb(3, riak3_unit, 0)
+
+        # remove one (no api atm, so directly to trigger notification)
         yield  self.client.delete(
             "/relations/%s/peer/%s" % (riak_states["relation"].internal_id,
                                        riak2_unit.internal_id))
-        yield wait_callback[4]
-        self.verify_unit_watch_result(
-            results[4],
-            old_units=[riak2_unit, riak3_unit], new_units=[riak3_unit])
+        yield checker.assert_members_cb(
+            4, [riak2_unit, riak3_unit], [riak3_unit])
 
         # modify one.
         yield riak3_relation.set_data(dict(later="eventually"))
-        yield wait_callback[5]
-        self.verify_unit_watch_result(
-            results[5], modified=((riak3_unit, 1),))
+        yield checker.assert_settings_cb(5, riak3_unit, 1)
 
     @inlineCallbacks
     def test_watch_role_container_created_concurrently(self):
@@ -1126,6 +1086,7 @@
             "mysql", "client-server", "", "server")
         wordpress_states = yield self.add_relation_service_unit_from_endpoints(
             wordpress_ep, mysql_ep)
+        wordpress_unit = wordpress_states["unit"]
         mysql_states = yield self.add_opposite_service_unit(
             wordpress_states)
 
@@ -1146,23 +1107,10 @@
 
         self.mocker.replay()
 
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        # Watch
-        watcher = yield mysql_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(mysql_states["unit_relation"])
         yield watcher.start()
-
-        # Verify
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[wordpress_states["unit"]])
+        yield checker.assert_members_cb(0, [], [wordpress_unit])
 
     @inlineCallbacks
     def test_watch_deleted_modify_notifications(self):
@@ -1178,48 +1126,31 @@
             "mysql", "client-server", "", "server")
         wordpress_states = yield self.add_relation_service_unit_from_endpoints(
             wordpress_ep, mysql_ep)
+        wordpress_unit = wordpress_states["unit"]
         mysql_states = yield self.add_opposite_service_unit(
             wordpress_states)
 
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
         # Start watching
-        watcher = yield mysql_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(mysql_states["unit_relation"])
         yield watcher.start()
-
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[wordpress_states["unit"]])
-
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((wordpress_states["unit"], 0),))
+        yield checker.assert_members_cb(0, [], [wordpress_unit])
+        yield checker.assert_settings_cb(1, wordpress_unit, 0)
 
         # Delete the presence path
         presence_path = "/relations/%s/client/%s" % (
             wordpress_states["relation"].internal_id,
             wordpress_states["unit"].internal_id)
         yield self.client.delete(presence_path)
+        yield checker.assert_members_cb(2, [wordpress_unit], [])
 
         # Modify the settings path
         settings_path = self.get_unit_settings_path(wordpress_states)
         yield self.client.set(settings_path, "some random string")
 
-        # Verify only deletion callback
-        yield wait_callback[2]
-        self.verify_unit_watch_result(
-            results[2], old_units=[wordpress_states["unit"]], new_units=[])
-
         # Give a moment to ensure we don't see any new callbacks
         yield self.sleep(0.1)
-        self.assertEquals(len(results), 3)
+        checker.assert_cb_count(3)
 
     @inlineCallbacks
     def test_watch_with_settings_deleted(self):
@@ -1236,29 +1167,15 @@
             "mysql", "client-server", "", "server")
         wordpress_states = yield self.add_relation_service_unit_from_endpoints(
             wordpress_ep, mysql_ep)
+        wordpress_unit = wordpress_states["unit"]
         mysql_states = yield self.add_opposite_service_unit(
             wordpress_states)
 
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        # Start watching
-        watcher = yield mysql_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(mysql_states["unit_relation"])
         yield watcher.start()
-
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[wordpress_states["unit"]])
-
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((wordpress_states["unit"], 0),))
+        yield checker.assert_members_cb(0, [], [wordpress_unit])
+        yield checker.assert_settings_cb(1, wordpress_unit, 0)
 
         # Delete the settings path
         settings_path = "/relations/%s/settings/%s" % (
@@ -1268,7 +1185,7 @@
 
         # Verify no callbacks
         yield self.sleep(0.1)
-        self.assertEqual(len(results), 2)
+        checker.assert_cb_count(2)
 
         # Recreate the settings path; this should trigger a callback.
         # Note that this is not likely to happen in reality, and if it does
@@ -1276,15 +1193,11 @@
         # to 0, and HookScheduler depends on that value continuing to increase
         # so it can determine whether changes happened while it was inactive.
         yield self.client.create(settings_path, "abc")
-        yield wait_callback[2]
-        self.verify_unit_watch_result(
-            results[2], modified=((wordpress_states["unit"], 0),))
+        yield checker.assert_settings_cb(2, wordpress_unit, 0)
 
         # And modify it.
         yield self.client.set(settings_path, "123")
-        yield wait_callback[3]
-        self.verify_unit_watch_result(
-            results[3], modified=((wordpress_states["unit"], 1),))
+        yield checker.assert_settings_cb(3, wordpress_unit, 1)
 
     @inlineCallbacks
     def test_watch_start_stop_start_with_existing_service(self):
@@ -1304,28 +1217,15 @@
             "mysql", "client-server", "", "server")
         wordpress_states = yield self.add_relation_service_unit_from_endpoints(
             wordpress_ep, mysql_ep)
+        wordpress_unit = wordpress_states["unit"]
         mysql_states = yield self.add_opposite_service_unit(
             wordpress_states)
 
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        # Start watching
-        watcher = yield mysql_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(mysql_states["unit_relation"])
         yield watcher.start()
-
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[wordpress_states["unit"]])
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((wordpress_states["unit"], 0),))
+        yield checker.assert_members_cb(0, [], [wordpress_unit])
+        yield checker.assert_settings_cb(1, wordpress_unit, 0)
 
         # Stop watching
         watcher.stop()
@@ -1333,26 +1233,20 @@
         # Add a new unit
         wordpress2_states = yield self.add_related_service_unit(
             wordpress_states)
+        wordpress2_unit = wordpress2_states["unit"]
 
-        # Modify a unit
+        # Modify a unit (this change will not be detected, ever)
         yield wordpress_states["unit_relation"].set_data(dict(hello="world"))
 
         # Verify no callbacks
         yield self.sleep(0.1)
-        self.assertEqual(len(results), 2)
+        checker.assert_cb_count(2)
 
-        # Start watching
+        # Start watching again; watch for addition
         yield watcher.start()
-
-        # Verify we see the addition.
-        yield wait_callback[2]
-        self.verify_unit_watch_result(
-            results[2],
-            old_units=[wordpress_states["unit"]],
-            new_units=[wordpress_states["unit"], wordpress2_states["unit"]])
-        yield wait_callback[3]
-        self.verify_unit_watch_result(
-            results[3], modified=((wordpress2_states["unit"], 0),))
+        yield checker.assert_members_cb(
+            2, [wordpress_unit], [wordpress_unit, wordpress2_unit])
+        yield checker.assert_settings_cb(3, wordpress2_unit, 0)
 
     @inlineCallbacks
     def test_watch_start_stop_start_with_new_service(self):
@@ -1373,48 +1267,37 @@
         mysql_states = yield self.add_relation_service_unit_from_endpoints(
             mysql_ep, wordpress_ep)
 
-        # setup callbacks and start observing.
-        wait_callback = [Deferred() for i in range(5)]
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-
-        # Start watching
-        watcher = yield mysql_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self)
+        watcher = yield checker.watch(mysql_states["unit_relation"])
         yield watcher.start()
-
-        # Stop watching
         watcher.stop()
 
-        # Add the new service and a unit
+        # Add the new service and 2 units
         wordpress_states = yield self.add_opposite_service_unit(
             mysql_states)
-
-        # Add another unit
         wordpress2_states = yield self.add_related_service_unit(
             wordpress_states)
+        wordpress_unit = wordpress_states["unit"]
+        wordpress2_unit = wordpress2_states["unit"]
 
         # Modify a unit
         yield wordpress_states["unit_relation"].set_data(dict(hello="world"))
 
         # Verify no callbacks
         yield self.sleep(0.1)
-        self.assertEqual(len(results), 0)
+        checker.assert_cb_count(0)
 
         # Start watching
         yield watcher.start()
-
-        # Wait a moment for callback
-        yield wait_callback[0]
-
-        self.assertEqual(len(results), 1)
-        self.verify_unit_watch_result(
-            results[0],
-            old_units=[],
-            new_units=[wordpress_states["unit"], wordpress2_states["unit"]])
+        yield checker.assert_members_cb(
+            0, [], [wordpress_unit, wordpress2_unit])
+
+        # We expect a settings callback for each...
+        yield checker.wait_for_cb(2)
+
+        # ...but only one
+        yield self.sleep(0.1)
+        checker.assert_cb_count(3)
 
     @inlineCallbacks
     def test_watch_user_callback_invocation_delays_node_watch(self):
@@ -1430,18 +1313,8 @@
         riak_states = yield self.add_relation_service_unit(
             "riak", "riak", "kvstore", "peer")
 
-        wait_callback = [Deferred() for i in range(10)]
-        finish_callback = [Deferred() for i in range(10)]
-
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-            return finish_callback[len(results)-1]
-
-        watcher = yield riak_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self, block_cbs=True)
+        watcher = yield checker.watch(riak_states["unit_relation"])
         yield watcher.start()
 
         # Create a new unit and add it to the relation.
@@ -1450,60 +1323,53 @@
             riak_unit2)
 
         # Wait for it
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[riak_unit2])
+        yield checker.assert_members_cb(0, [], [riak_unit2])
 
         # We are also expecting a notification for the initial settings version
         # ...but we won't get that until the first callback is done
         self.sleep(0.1)
-        self.assertEquals(len(results), 1)
+        checker.assert_cb_count(1)
 
         # While we wait for this, someone modifies the settings
         yield riak_unit2_rel.set_data(dict(hello="world"))
 
         # Hey, the add callback finished!
-        finish_callback[0].callback(True)
+        checker.unblock_cb(0)
 
         # OK, now we expect to see the initial setting version callback
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((riak_unit2, 0),))
+        yield checker.assert_settings_cb(1, riak_unit2, 0)
 
         # ...but that is also taking a long time, so we shouldn't expect to see
         # the callback for the explicit modification yet...
-        self.sleep(0.1)
-        self.assertEquals(len(results), 2)
+        yield self.sleep(0.1)
+        checker.assert_cb_count(2)
 
         # ...or, in fact, for this other modification that just happened, which
         # will be collapsed into the other one...
         yield riak_unit2_rel.set_data(dict(hello="world 2"))
-        self.sleep(0.1)
-        self.assertEquals(len(results), 2)
+        yield self.sleep(0.1)
+        checker.assert_cb_count(2)
 
         # ...so, we should have 1 callback in progress, and only 1 pending
         # notification. OK, finish the callback...
-        finish_callback[1].callback(True)
+        checker.unblock_cb(1)
 
         # ...and wait for the change notification.
-        yield wait_callback[2]
-        self.assertEqual(len(results), 3)
-        self.verify_unit_watch_result(
-            results[2], modified=((riak_unit2, 2),))
+        yield checker.assert_settings_cb(2, riak_unit2, 2)
 
         # Finish the callback and verify no other invocations.
-        finish_callback[2].callback(True)
+        checker.unblock_cb(2)
         yield self.sleep(0.1)
-        self.assertEqual(len(results), 3)
+        checker.assert_cb_count(3)
 
-        # Modify the node again, we should see this change, immediately.
+        # Modify the node again; we should see this change immediately.
         yield riak_unit2_rel.set_data(dict(hello="goodbye"))
-        yield wait_callback[3]
-        finish_callback[3].callback(True)
+        yield checker.assert_settings_cb(3, riak_unit2, 3)
 
-        self.verify_unit_watch_result(
-            results[3], modified=((riak_unit2, 3),))
-        self.assertEqual(len(results), 4)
+        # And again, finish the callback and verify no other invocations.
+        checker.unblock_cb(3)
+        yield self.sleep(0.1)
+        checker.assert_cb_count(4)
 
         node_path = "/relations/relation-0000000000/settings/unit-0000000001"
         expected_output = (
@@ -1527,18 +1393,8 @@
         riak_states = yield self.add_relation_service_unit(
             "riak", "riak", "kvstore", "peer")
 
-        wait_callback = [Deferred() for i in range(10)]
-        finish_callback = [Deferred() for i in range(10)]
-
-        results = []
-
-        def watch_related(old_units=None, new_units=None, modified=None):
-            results.append((old_units, new_units, modified))
-            wait_callback[len(results)-1].callback(True)
-            return finish_callback[len(results)-1]
-
-        watcher = yield riak_states["unit_relation"].watch_related_units(
-            watch_related)
+        checker = WatchChecker(self, block_cbs=True)
+        watcher = yield checker.watch(riak_states["unit_relation"])
         yield watcher.start()
 
         # Create a new unit and add it to the relation.
@@ -1547,9 +1403,7 @@
             riak_unit2)
 
         # Wait for it
-        yield wait_callback[0]
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[riak_unit2])
+        yield checker.assert_members_cb(0, [], [riak_unit2])
 
         # Now add a new unit: we won't see it immediately, since the callback
         # is still executing, but we will have a container change pending
@@ -1557,12 +1411,10 @@
         yield riak_states["service_relation"].add_unit_state(
             riak_unit3)
 
-        # Finish the first callback, immediately hit the settings version
+        # Finish the first callback; immediately hit the settings version
         # callback, which will also take a while
-        finish_callback[0].callback(True)
-        yield wait_callback[1]
-        self.verify_unit_watch_result(
-            results[1], modified=((riak_unit2, 0),))
+        checker.unblock_cb(0)
+        yield checker.assert_settings_cb(1, riak_unit2, 0)
 
         # Adding another unit; will be rolled into the container change we're
         # already expecting from before
@@ -1572,12 +1424,9 @@
 
         # Now release the container callback, and verify the callback
         # for both the new nodes.
-        finish_callback[1].callback(True)
-        yield wait_callback[2]
-        self.verify_unit_watch_result(
-            results[2],
-            old_units=[riak_unit2],
-            new_units=[riak_unit2, riak_unit3, riak_unit4])
+        checker.unblock_cb(1)
+        yield checker.assert_members_cb(
+            2, [riak_unit2], [riak_unit2, riak_unit3, riak_unit4])
 
     @inlineCallbacks
     def test_watch_concurrent_callback_execution(self):
@@ -1585,35 +1434,6 @@
 
         IFF they are not synchronous and not on the same node.
         """
-        #Setup parallel callbacks bookeeping and implementation
-        wait_callback = dict([(i, Deferred()) for i in range(10)])
-        results = {}
-        success_deferrals = {
-            2: Deferred(),
-            5: Deferred()}
-
-        @inlineCallbacks
-        def watch_related(
-            old_units=None, new_units=None, modified=None, callback_id=None):
-            success_result = success_deferrals.get(callback_id)
-            if success_result is not None:
-                yield success_result
-            results[callback_id] = (old_units, new_units, modified)
-            wait_callback[callback_id].callback(True)
-
-        class ParallelWatcherCallback(object):
-
-            def __init__(self):
-                self.callback_id_sequence = 0
-
-            def __call__(self, *args, **kw):
-                callback = functools.partial(
-                    watch_related, callback_id=self.callback_id_sequence)
-                self.callback_id_sequence += 1
-                return callback(*args, **kw)
-
-        watcher_callback = ParallelWatcherCallback()
-
         # Add the relation, services, and related units.
         wordpress_ep = RelationEndpoint(
             "wordpress", "client-server", "", "client")
@@ -1621,56 +1441,74 @@
             "mysql", "client-server", "", "server")
         wordpress_states = yield self.add_relation_service_unit_from_endpoints(
             wordpress_ep, mysql_ep)
+        wordpress_unit = wordpress_states["unit"]
         mysql_states = yield self.add_opposite_service_unit(
             wordpress_states)
 
-        # Start watching, and give a moment to establish
-        watcher = yield mysql_states["unit_relation"].watch_related_units(
-            watcher_callback)
+        checker = WatchChecker(self, block_cbs=True)
+        # To verify parallel execution, this checker will make us wait for
+        # some of the callbacks (2 and 5), but leave the rest unimpeded.
+        for i in (0, 1, 3, 4, 6):
+            checker.unblock_cb(i)
+
+        watcher = yield checker.watch(mysql_states["unit_relation"])
         yield watcher.start()
+        yield checker.wait_for_cb(0)
+        yield checker.wait_for_cb(1)
 
-        # Modify a unit
+        # Modify a unit (blocking callback)
         yield wordpress_states["unit_relation"].set_data(dict(hello="world"))
+        yield checker.wait_for_cb(2)
+
+        # Modify the unit again (will wait for previous)
+        yield wordpress_states["unit_relation"].set_data(dict(hello="world 2"))
 
         # Add a unit.
         wordpress2_states = yield self.add_related_service_unit(
             wordpress_states)
+        wordpress2_unit = wordpress2_states["unit"]
+        yield checker.wait_for_cb(3)
+        yield checker.wait_for_cb(4)
 
-        # Delete a unit
+        # Delete a unit (blocking callbck)
         presence_path = "/relations/%s/client/%s" % (
             wordpress_states["relation"].internal_id,
             wordpress_states["unit"].internal_id)
         yield self.client.delete(presence_path)
-
-        # Verify the parallel execution, give a moment for callbacks to finish.
-        yield self.sleep(0.1)
-        self.assertEqual(len(results), 4)
-
-        self.verify_unit_watch_result(
-            results[0], old_units=[], new_units=[wordpress_states["unit"]])
-        self.verify_unit_watch_result(
-            results[1], modified=((wordpress_states["unit"], 0),))
-        self.verify_unit_watch_result(
-            results[3],
-            old_units=[wordpress_states["unit"]],
-            new_units=[wordpress_states["unit"], wordpress2_states["unit"]])
-        self.verify_unit_watch_result(
-            results[4], modified=((wordpress2_states["unit"], 0),))
-
-        for k, v in success_deferrals.items():
-            v.callback(True)
-
-        # Verify all callback results, give a moment for them to finish.
-        yield wait_callback[2]
-        yield wait_callback[5]
-        self.assertEquals(len(results), 6)
-
-        self.verify_unit_watch_result(
-            results[2], modified=((wordpress_states["unit"], 1),))
-        self.verify_unit_watch_result(
-            results[5],
-            old_units=[wordpress_states["unit"], wordpress2_states["unit"]],
-            new_units=[wordpress2_states["unit"]])
+        yield checker.wait_for_cb(5)
+
+        # ...and delete the other unit (also blocked)
+        presence_path = "/relations/%s/client/%s" % (
+            wordpress2_states["relation"].internal_id,
+            wordpress2_states["unit"].internal_id)
+        yield self.client.delete(presence_path)
+
+        # Verify that all unblocked callbacks have started correctly
+        yield checker.assert_members_cb(0, [], [wordpress_unit])
+        yield checker.assert_settings_cb(1, wordpress_unit, 0)
+        yield checker.assert_settings_cb(2, wordpress_unit, 1)
+        yield checker.assert_members_cb(
+            3, [wordpress_unit], [wordpress_unit, wordpress2_unit])
+        yield checker.assert_settings_cb(4, wordpress2_unit, 0)
+        yield checker.assert_members_cb(
+            5, [wordpress_unit, wordpress2_unit], [wordpress2_unit])
+        checker.assert_cb_count(6)
+        # OK, fine, but cbs 2 and 5 are still blocking.
+
+        # Whoops, looks like the unit got deleted before we could notify the
+        # second settings change. Check nothing happens:
+        checker.unblock_cb(2)
+        yield self.sleep(0.1)
+        checker.assert_cb_count(6)
+
+        # Now complete processing of the first delete, and check that we then
+        # *do* get notified of the second delete:
+        checker.unblock_cb(5)
+        yield checker.assert_members_cb(6, [wordpress2_unit], [])
+
+        # ...and finally double-check no further callbacks:
+        yield self.sleep(0.1)
+        checker.assert_cb_count(7)
 
     @inlineCallbacks
     def test_watch_unknown_relation_role_error(self):
@@ -1685,5 +1523,6 @@
             self.fail("Should not be called")
 
         yield self.failUnlessFailure(
-            wordpress_states["unit_relation"].watch_related_units(not_called),
+            wordpress_states["unit_relation"].watch_related_units(
+                not_called, not_called),
             UnknownRelationRole)

=== modified file 'juju/unit/lifecycle.py'
--- juju/unit/lifecycle.py	2011-12-05 02:11:54 +0000
+++ juju/unit/lifecycle.py	2011-12-05 02:11:55 +0000
@@ -436,7 +436,8 @@
             # Create a watcher if we don't have one yet.
             if self._watcher is None:
                 self._watcher = yield self._unit_relation.watch_related_units(
-                    self._scheduler.notify_change)
+                    self._scheduler.members_changed,
+                    self._scheduler.settings_changed)
             # And start the watcher.
             if watches:
                 yield self._watcher.start()