func (s *PeekerSuite) TestUpdatesEmptyQueueSpam(c *gc.C) {
	source := hooktesting.NewEmptySource()
	defer statetesting.AssertStop(c, source)

	peeker := hook.NewPeeker(source)
	defer statetesting.AssertStop(c, peeker)

	// Spam all channels continuously for a bit.
	timeout := time.After(coretesting.LongWait)
	changeCount := 0
	updateCount := 0
	for i := 0; i < 100; i++ {
		select {
		case peek, ok := <-peeker.Peeks():
			c.Fatalf("got unexpected peek: %#v %#v", peek, ok)
		case source.ChangesC <- source.NewChange("!"):
			changeCount++
		case update, ok := <-source.UpdatesC:
			c.Assert(ok, jc.IsTrue)
			c.Assert(update, gc.Equals, "!")
			updateCount++
		case <-timeout:
			c.Fatalf("not enough things happened in time")
		}
	}

	// Check sane end state.
	c.Check(changeCount, gc.Equals, 50)
	c.Check(updateCount, gc.Equals, 50)
}

func (s *deployerSuite) TestWatchUnits(c *gc.C) {
	// TODO(dfc) fix state.Machine to return a MachineTag
	machine, err := s.st.Machine(s.machine.Tag().(names.MachineTag))
	c.Assert(err, gc.IsNil)
	w, err := machine.WatchUnits()
	c.Assert(err, gc.IsNil)
	defer statetesting.AssertStop(c, w)
	wc := statetesting.NewStringsWatcherC(c, s.BackingState, w)

	// Initial event.
	wc.AssertChange("mysql/0", "logging/0")
	wc.AssertNoChange()

	// Change something other than the lifecycle and make sure it's
	// not detected.
	err = s.subordinate.SetPassword("foo")
	c.Assert(err, gc.ErrorMatches, "password is only 3 bytes long, and is not a valid Agent password")
	wc.AssertNoChange()

	err = s.subordinate.SetPassword("foo-12345678901234567890")
	c.Assert(err, gc.IsNil)
	wc.AssertNoChange()

	// Make the subordinate dead and check it's detected.
	err = s.subordinate.EnsureDead()
	c.Assert(err, gc.IsNil)
	wc.AssertChange("logging/0")
	wc.AssertNoChange()

	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *unitSuite) TestWatchMeterStatus(c *gc.C) {
	w, err := s.apiUnit.WatchMeterStatus()
	defer statetesting.AssertStop(c, w)
	wc := statetesting.NewNotifyWatcherC(c, s.BackingState, w)

	// Initial event.
	wc.AssertOneChange()

	err = s.wordpressUnit.SetMeterStatus("GREEN", "ok")
	c.Assert(err, jc.ErrorIsNil)
	err = s.wordpressUnit.SetMeterStatus("AMBER", "ok")
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertOneChange()

	// Non-change is not reported.
	err = s.wordpressUnit.SetMeterStatus("AMBER", "ok")
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertNoChange()

	mm, err := s.State.MetricsManager()
	c.Assert(err, jc.ErrorIsNil)
	err = mm.SetLastSuccessfulSend(time.Now())
	c.Assert(err, jc.ErrorIsNil)
	for i := 0; i < 3; i++ {
		err := mm.IncrementConsecutiveErrors()
		c.Assert(err, jc.ErrorIsNil)
	}
	status := mm.MeterStatus()
	c.Assert(status.Code, gc.Equals, state.MeterAmber) // Confirm meter status has changed
	wc.AssertOneChange()

	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *unitSuite) TestWatchActionNotifications(c *gc.C) {
	w, err := s.apiUnit.WatchActionNotifications()
	c.Assert(err, jc.ErrorIsNil)

	defer statetesting.AssertStop(c, w)
	wc := statetesting.NewStringsWatcherC(c, s.BackingState, w)

	// Initial event.
	wc.AssertChange()

	// Add a couple of actions and make sure the changes are detected.
	action, err := s.wordpressUnit.AddAction("fakeaction", map[string]interface{}{
		"outfile": "foo.txt",
	})
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertChange(action.Id())

	action, err = s.wordpressUnit.AddAction("fakeaction", map[string]interface{}{
		"outfile": "foo.bz2",
		"compression": map[string]interface{}{
			"kind":    "bzip",
			"quality": float64(5.0),
		},
	})
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertChange(action.Id())

	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *PeekerSuite) TestPeeksBlockUntilConsumed(c *gc.C) {
	source := hooktesting.NewFullBufferedSource()
	defer statetesting.AssertStop(c, source)

	peeker := hook.NewPeeker(source)
	defer statetesting.AssertStop(c, peeker)

	// Collect a peek...
	timeout := time.After(coretesting.LongWait)
	select {
	case <-timeout:
		c.Fatalf("failed to receive peek")
	case peek, ok := <-peeker.Peeks():
		c.Assert(ok, jc.IsTrue)
		c.Assert(peek.HookInfo(), gc.Equals, hook.Info{Kind: hooks.Install})

		// ...and check that changes can't be delivered...
		select {
		case source.ChangesC <- source.NewChange(nil):
			c.Fatalf("delivered change while supposedly peeking")
		default:
		}

		// ...before the peek is consumed, at which point changes are unblocked.
		peek.Consume()
		select {
		case source.ChangesC <- source.NewChange(nil):
		case <-timeout:
			c.Fatalf("failed to unblock changes")
		}
	}
}

func (s *serviceSuite) TestWatchRelations(c *gc.C) {
	w, err := s.apiService.WatchRelations()
	c.Assert(err, jc.ErrorIsNil)
	defer statetesting.AssertStop(c, w)
	wc := statetesting.NewStringsWatcherC(c, s.BackingState, w)

	// Initial event.
	wc.AssertChange()
	wc.AssertNoChange()

	// Change something other than the lifecycle and make sure it's
	// not detected.
	err = s.wordpressService.SetExposed()
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertNoChange()

	// Add another service and relate it to wordpress,
	// check it's detected.
	s.addMachineServiceCharmAndUnit(c, "mysql")
	rel := s.addRelation(c, "wordpress", "mysql")
	wc.AssertChange(rel.String())

	// Destroy the relation and check it's detected.
	err = rel.Destroy()
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertChange(rel.String())
	wc.AssertNoChange()

	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *provisionerSuite) TestWatchEnvironMachines(c *gc.C) {
	w, err := s.provisioner.WatchEnvironMachines()
	c.Assert(err, gc.IsNil)
	defer statetesting.AssertStop(c, w)
	wc := statetesting.NewStringsWatcherC(c, s.BackingState, w)

	// Initial event.
	wc.AssertChange(s.machine.Id())

	// Add another 2 machines make sure they are detected.
	otherMachine, err := s.State.AddMachine("quantal", state.JobHostUnits)
	c.Assert(err, gc.IsNil)
	otherMachine, err = s.State.AddMachine("quantal", state.JobHostUnits)
	c.Assert(err, gc.IsNil)
	wc.AssertChange("1", "2")

	// Change the lifecycle of last machine.
	err = otherMachine.EnsureDead()
	c.Assert(err, gc.IsNil)
	wc.AssertChange("2")

	// Add a container and make sure it's not detected.
	template := state.MachineTemplate{
		Series: "quantal",
		Jobs:   []state.MachineJob{state.JobHostUnits},
	}
	_, err = s.State.AddMachineInsideMachine(template, s.machine.Id(), instance.LXC)
	c.Assert(err, gc.IsNil)
	wc.AssertNoChange()

	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *RelationUnitSuite) TestPeerWatchScope(c *gc.C) {
	pr := NewPeerRelation(c, s.State, s.Owner)

	// Test empty initial event.
	w0 := pr.ru0.WatchScope()
	defer testing.AssertStop(c, w0)
	s.assertScopeChange(c, w0, nil, nil)
	s.assertNoScopeChange(c, w0)

	// ru0 enters; check no change, but settings written.
	assertNotInScope(c, pr.ru0)
	err := pr.ru0.EnterScope(map[string]interface{}{"foo": "bar"})
	c.Assert(err, jc.ErrorIsNil)
	s.assertNoScopeChange(c, w0)
	node, err := pr.ru0.Settings()
	c.Assert(err, jc.ErrorIsNil)
	c.Assert(node.Map(), gc.DeepEquals, map[string]interface{}{"foo": "bar"})
	assertJoined(c, pr.ru0)

	// ru1 enters; check change is observed.
	assertNotInScope(c, pr.ru1)
	err = pr.ru1.EnterScope(nil)
	c.Assert(err, jc.ErrorIsNil)
	s.assertScopeChange(c, w0, []string{"riak/1"}, nil)
	s.assertNoScopeChange(c, w0)
	assertJoined(c, pr.ru1)

	// ru1 enters again, check no problems and no changes.
	err = pr.ru1.EnterScope(nil)
	c.Assert(err, jc.ErrorIsNil)
	s.assertNoScopeChange(c, w0)
	assertJoined(c, pr.ru1)

	// Stop watching; ru2 enters.
	testing.AssertStop(c, w0)
	assertNotInScope(c, pr.ru2)
	err = pr.ru2.EnterScope(nil)
	c.Assert(err, jc.ErrorIsNil)
	assertJoined(c, pr.ru2)

	// Start watch again, check initial event.
	w0 = pr.ru0.WatchScope()
	defer testing.AssertStop(c, w0)
	s.assertScopeChange(c, w0, []string{"riak/1", "riak/2"}, nil)
	s.assertNoScopeChange(c, w0)

	// ru1 leaves; check event.
	assertJoined(c, pr.ru1)
	err = pr.ru1.LeaveScope()
	c.Assert(err, jc.ErrorIsNil)
	s.assertScopeChange(c, w0, nil, []string{"riak/1"})
	s.assertNoScopeChange(c, w0)
	assertNotInScope(c, pr.ru1)

	// ru1 leaves again; check no problems and no changes.
	err = pr.ru1.LeaveScope()
	c.Assert(err, jc.ErrorIsNil)
	s.assertNoScopeChange(c, w0)
	assertNotInScope(c, pr.ru1)
}

func (s *UnitSuite) TestWatchUnit(c *gc.C) {
	preventUnitDestroyRemove(c, s.unit)
	w := s.unit.Watch()
	defer testing.AssertStop(c, w)

	// Initial event.
	wc := testing.NewNotifyWatcherC(c, s.State, w)
	wc.AssertOneChange()

	// Make one change (to a separate instance), check one event.
	unit, err := s.State.Unit(s.unit.Name())
	c.Assert(err, jc.ErrorIsNil)
	s.setAssignedMachineAddresses(c, unit)
	wc.AssertOneChange()

	// Make two changes, check one event.
	err = unit.SetPassword("arble-farble-dying-yarble")
	c.Assert(err, jc.ErrorIsNil)
	err = unit.Destroy()
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertOneChange()

	// Stop, check closed.
	testing.AssertStop(c, w)
	wc.AssertClosed()

	// Remove unit, start new watch, check single event.
	err = unit.EnsureDead()
	c.Assert(err, jc.ErrorIsNil)
	err = unit.Remove()
	c.Assert(err, jc.ErrorIsNil)
	w = s.unit.Watch()
	defer testing.AssertStop(c, w)
	testing.NewNotifyWatcherC(c, s.State, w).AssertOneChange()
}

func (s *HookSenderSuite) TestHandlesUpdatesEmptyQueueSpam(c *gc.C) {
	source := hooktesting.NewEmptySource()
	defer statetesting.AssertStop(c, source)

	out := make(chan hook.Info)
	sender := hook.NewSender(out, source)
	defer statetesting.AssertStop(c, sender)

	// Spam all channels continuously for a bit.
	timeout := time.After(coretesting.LongWait)
	changeCount := 0
	updateCount := 0
	for i := 0; i < 100; i++ {
		select {
		case hi, ok := <-out:
			c.Fatalf("got unexpected hook: %#v %#v", hi, ok)
		case source.ChangesC <- source.NewChange("sent"):
			changeCount++
		case update, ok := <-source.UpdatesC:
			c.Assert(ok, jc.IsTrue)
			c.Assert(update, gc.Equals, "sent")
			updateCount++
		case <-timeout:
			c.Fatalf("not enough things happened in time")
		}
	}

	// Check sane end state.
	c.Check(changeCount, gc.Equals, 50)
	c.Check(updateCount, gc.Equals, 50)
}

func (s *UpgradeSuite) TestWatch(c *gc.C) {
	v111 := vers("1.1.1")
	v123 := vers("1.2.3")
	serverIdB, serverIdC := s.addControllers(c)
	s.provision(c, serverIdB, serverIdC)

	w := s.State.WatchUpgradeInfo()
	defer statetesting.AssertStop(c, w)

	// initial event
	wc := statetesting.NewNotifyWatcherC(c, s.State, w)
	wc.AssertOneChange()

	// single change is reported
	_, err := s.State.EnsureUpgradeInfo(s.serverIdA, v111, v123)
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertOneChange()

	// non-change is not reported
	_, err = s.State.EnsureUpgradeInfo(s.serverIdA, v111, v123)
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertNoChange()

	// changes are coalesced
	_, err = s.State.EnsureUpgradeInfo(serverIdB, v111, v123)
	c.Assert(err, jc.ErrorIsNil)
	_, err = s.State.EnsureUpgradeInfo(serverIdC, v111, v123)
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertOneChange()

	// closed on stop
	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *rebootSuite) TearDownTest(c *gc.C) {
	if s.machine.resources != nil {
		s.machine.resources.StopAll()
	}
	if s.machine.w != nil {
		statetesting.AssertStop(c, s.machine.w)
		s.machine.wc.AssertClosed()
	}

	if s.container.resources != nil {
		s.container.resources.StopAll()
	}
	if s.container.w != nil {
		statetesting.AssertStop(c, s.container.w)
		s.container.wc.AssertClosed()
	}

	if s.nestedContainer.resources != nil {
		s.nestedContainer.resources.StopAll()
	}
	if s.nestedContainer.w != nil {
		statetesting.AssertStop(c, s.nestedContainer.w)
		s.nestedContainer.wc.AssertClosed()
	}
	s.JujuConnSuite.TearDownTest(c)
}

func (s *APIAddresserTests) TestWatchAPIHostPorts(c *gc.C) {
	expectServerAddrs := [][]network.HostPort{{{
		Address: network.NewAddress("0.1.2.3", network.ScopeUnknown),
		Port:    1234,
	}}}
	err := s.state.SetAPIHostPorts(expectServerAddrs)
	c.Assert(err, gc.IsNil)

	w, err := s.facade.WatchAPIHostPorts()
	c.Assert(err, gc.IsNil)
	defer statetesting.AssertStop(c, w)

	wc := statetesting.NewNotifyWatcherC(c, s.state, w)

	// Initial event.
	wc.AssertOneChange()

	// Change the state addresses and check that we get a notification
	expectServerAddrs[0][0].Value = "0.1.99.99"

	err = s.state.SetAPIHostPorts(expectServerAddrs)
	c.Assert(err, gc.IsNil)

	wc.AssertOneChange()

	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *ServiceSuite) TestWatchService(c *gc.C) {
	w := s.mysql.Watch()
	defer testing.AssertStop(c, w)

	// Initial event.
	wc := testing.NewNotifyWatcherC(c, s.State, w)
	wc.AssertOneChange()

	// Make one change (to a separate instance), check one event.
	service, err := s.State.Service(s.mysql.Name())
	c.Assert(err, gc.IsNil)
	err = service.SetExposed()
	c.Assert(err, gc.IsNil)
	wc.AssertOneChange()

	// Make two changes, check one event.
	err = service.ClearExposed()
	c.Assert(err, gc.IsNil)
	err = service.SetCharm(s.charm, true)
	c.Assert(err, gc.IsNil)
	wc.AssertOneChange()

	// Stop, check closed.
	testing.AssertStop(c, w)
	wc.AssertClosed()

	// Remove service, start new watch, check single event.
	err = service.Destroy()
	c.Assert(err, gc.IsNil)
	w = s.mysql.Watch()
	defer testing.AssertStop(c, w)
	testing.NewNotifyWatcherC(c, s.State, w).AssertOneChange()
}

func (s *FilterSuite) TestConfigEvents(c *gc.C) {
	f, err := filter.NewFilter(s.uniter, s.unit.Tag().(names.UnitTag))
	c.Assert(err, jc.ErrorIsNil)
	defer statetesting.AssertStop(c, f)

	err = s.machine.SetProviderAddresses(network.NewAddress("0.1.2.3"))
	c.Assert(err, jc.ErrorIsNil)

	// Test no changes before the charm URL is set.
	configC := s.notifyAsserterC(c, f.ConfigEvents())
	configC.AssertNoReceive()

	// Set the charm URL to trigger config events.
	err = f.SetCharm(s.wpcharm.URL())
	c.Assert(err, jc.ErrorIsNil)
	s.EvilSync()
	configC.AssertOneReceive()

	// Change the config; new event received.
	changeConfig := func(title interface{}) {
		err := s.wordpress.UpdateConfigSettings(charm.Settings{
			"blog-title": title,
		})
		c.Assert(err, jc.ErrorIsNil)
	}
	changeConfig("20,000 leagues in the cloud")
	configC.AssertOneReceive()

	// Change the config a few more times, then reset the events. We sync to
	// make sure the events have arrived in the watcher -- and then wait a
	// little longer, to allow for the delay while the events are coalesced
	// -- before we tell it to discard all received events. This would be
	// much better tested by controlling a mocked-out watcher directly, but
	// that's a bit inconvenient for this change.
	changeConfig(nil)
	changeConfig("the curious incident of the dog in the cloud")
	s.EvilSync()
	f.DiscardConfigEvent()
	configC.AssertNoReceive()

	// Change the addresses of the unit's assigned machine; new event received.
	err = s.machine.SetProviderAddresses(network.NewAddress("0.1.2.4"))
	c.Assert(err, jc.ErrorIsNil)
	s.BackingState.StartSync()
	configC.AssertOneReceive()

	// Check that a filter's initial event works with DiscardConfigEvent
	// as expected.
	f, err = filter.NewFilter(s.uniter, s.unit.Tag().(names.UnitTag))
	c.Assert(err, jc.ErrorIsNil)
	defer statetesting.AssertStop(c, f)
	s.BackingState.StartSync()
	f.DiscardConfigEvent()
	configC.AssertNoReceive()

	// Further changes are still collapsed as appropriate.
	changeConfig("forsooth")
	changeConfig("imagination failure")
	configC.AssertOneReceive()
}

func (s *machineSuite) TestWatchUnits(c *gc.C) {
	w, err := s.apiMachine.WatchUnits()
	c.Assert(err, gc.IsNil)
	defer statetesting.AssertStop(c, w)
	wc := statetesting.NewStringsWatcherC(c, s.BackingState, w)

	// Initial event.
	wc.AssertChange("wordpress/0")
	wc.AssertNoChange()

	// Change something other than the life cycle and make sure it's
	// not detected.
	err = s.machines[0].SetPassword("foo")
	c.Assert(err, gc.ErrorMatches, "password is only 3 bytes long, and is not a valid Agent password")
	wc.AssertNoChange()

	err = s.machines[0].SetPassword("foo-12345678901234567890")
	c.Assert(err, gc.IsNil)
	wc.AssertNoChange()

	// Unassign unit 0 from the machine and check it's detected.
	err = s.units[0].UnassignFromMachine()
	c.Assert(err, gc.IsNil)
	wc.AssertChange("wordpress/0")
	wc.AssertNoChange()

	statetesting.AssertStop(c, w)
	wc.AssertClosed()
}

func (s *provisionerSuite) TestWatchFilesystemAttachments(c *gc.C) {
	s.setupFilesystems(c)
	c.Assert(s.resources.Count(), gc.Equals, 0)

	args := params.Entities{Entities: []params.Entity{
		{"machine-0"},
		{s.State.ModelTag().String()},
		{"environ-adb650da-b77b-4ee8-9cbb-d57a9a592847"},
		{"machine-1"},
		{"machine-42"}},
	}
	result, err := s.api.WatchFilesystemAttachments(args)
	c.Assert(err, jc.ErrorIsNil)
	sort.Sort(byMachineAndEntity(result.Results[0].Changes))
	sort.Sort(byMachineAndEntity(result.Results[1].Changes))
	c.Assert(result, jc.DeepEquals, params.MachineStorageIdsWatchResults{
		Results: []params.MachineStorageIdsWatchResult{
			{
				MachineStorageIdsWatcherId: "1",
				Changes: []params.MachineStorageId{{
					MachineTag:    "machine-0",
					AttachmentTag: "filesystem-0-0",
				}},
			},
			{
				MachineStorageIdsWatcherId: "2",
				Changes: []params.MachineStorageId{{
					MachineTag:    "machine-0",
					AttachmentTag: "filesystem-1",
				}, {
					MachineTag:    "machine-0",
					AttachmentTag: "filesystem-2",
				}, {
					MachineTag:    "machine-2",
					AttachmentTag: "filesystem-3",
				}},
			},
			{Error: apiservertesting.ErrUnauthorized},
			{Error: apiservertesting.ErrUnauthorized},
			{Error: apiservertesting.ErrUnauthorized},
		},
	})

	// Verify the resources were registered and stop them when done.
	c.Assert(s.resources.Count(), gc.Equals, 2)
	v0Watcher := s.resources.Get("1")
	defer statetesting.AssertStop(c, v0Watcher)
	v1Watcher := s.resources.Get("2")
	defer statetesting.AssertStop(c, v1Watcher)

	// Check that the Watch has consumed the initial events ("returned" in
	// the Watch call)
	wc := statetesting.NewStringsWatcherC(c, s.State, v0Watcher.(state.StringsWatcher))
	wc.AssertNoChange()
	wc = statetesting.NewStringsWatcherC(c, s.State, v1Watcher.(state.StringsWatcher))
	wc.AssertNoChange()
}

func (s *InstancePollerSuite) TestWatchModelMachinesSuccess(c *gc.C) {
	// Add a couple of machines.
	s.st.SetMachineInfo(c, machineInfo{id: "2"})
	s.st.SetMachineInfo(c, machineInfo{id: "1"})

	expectedResult := params.StringsWatchResult{
		Error:            nil,
		StringsWatcherId: "1",
		Changes:          []string{"1", "2"}, // initial event (sorted ids)
	}
	result, err := s.api.WatchModelMachines()
	c.Assert(err, jc.ErrorIsNil)
	c.Assert(result, jc.DeepEquals, expectedResult)

	// Verify the watcher resource was registered.
	c.Assert(s.resources.Count(), gc.Equals, 1)
	resource1 := s.resources.Get("1")
	defer func() {
		if resource1 != nil {
			statetesting.AssertStop(c, resource1)
		}
	}()

	// Check that the watcher has consumed the initial event
	wc1 := statetesting.NewStringsWatcherC(c, s.st, resource1.(state.StringsWatcher))
	wc1.AssertNoChange()

	s.st.CheckCallNames(c, "WatchModelMachines")

	// Add another watcher to verify events coalescence.
	result, err = s.api.WatchModelMachines()
	c.Assert(err, jc.ErrorIsNil)
	expectedResult.StringsWatcherId = "2"
	c.Assert(result, jc.DeepEquals, expectedResult)
	s.st.CheckCallNames(c, "WatchModelMachines", "WatchModelMachines")
	c.Assert(s.resources.Count(), gc.Equals, 2)
	resource2 := s.resources.Get("2")
	defer statetesting.AssertStop(c, resource2)
	wc2 := statetesting.NewStringsWatcherC(c, s.st, resource2.(state.StringsWatcher))
	wc2.AssertNoChange()

	// Remove machine 1, check it's reported.
	s.st.RemoveMachine(c, "1")
	wc1.AssertChangeInSingleEvent("1")

	// Make separate changes, check they're combined.
	s.st.SetMachineInfo(c, machineInfo{id: "2", life: state.Dying})
	s.st.SetMachineInfo(c, machineInfo{id: "3"})
	s.st.RemoveMachine(c, "42") // ignored
	wc1.AssertChangeInSingleEvent("2", "3")
	wc2.AssertChangeInSingleEvent("1", "2", "3")

	// Stop the first watcher and assert its changes chan is closed.
	c.Assert(resource1.Stop(), jc.ErrorIsNil)
	wc1.AssertClosed()
	resource1 = nil
}

func (s *HookSenderSuite) TestStopsHooks(c *gc.C) {
	expect := hooktesting.HookList(hooks.Install, hooks.ConfigChanged, hooks.Start)
	source := hook.NewListSource(expect)
	out := make(chan hook.Info)
	sender := hook.NewSender(out, source)
	defer statetesting.AssertStop(c, sender)

	assertNext(c, out, expect[0])
	assertNext(c, out, expect[1])
	statetesting.AssertStop(c, sender)
	assertEmpty(c, out)
	c.Assert(source.Next(), gc.Equals, expect[2])
}

func (s *ActionSuite) TestUnitWatchActionNotifications(c *gc.C) {
	// get units
	unit1, err := s.State.Unit(s.unit.Name())
	c.Assert(err, jc.ErrorIsNil)
	preventUnitDestroyRemove(c, unit1)

	unit2, err := s.State.Unit(s.unit2.Name())
	c.Assert(err, jc.ErrorIsNil)
	preventUnitDestroyRemove(c, unit2)

	// queue some actions before starting the watcher
	fa1, err := unit1.AddAction("snapshot", nil)
	c.Assert(err, jc.ErrorIsNil)
	fa2, err := unit1.AddAction("snapshot", nil)
	c.Assert(err, jc.ErrorIsNil)

	// set up watcher on first unit
	w := unit1.WatchActionNotifications()
	defer statetesting.AssertStop(c, w)
	wc := statetesting.NewStringsWatcherC(c, s.State, w)
	// make sure the previously pending actions are sent on the watcher
	expect := expectActionIds(fa1, fa2)
	wc.AssertChange(expect...)
	wc.AssertNoChange()

	// add watcher on unit2
	w2 := unit2.WatchActionNotifications()
	defer statetesting.AssertStop(c, w2)
	wc2 := statetesting.NewStringsWatcherC(c, s.State, w2)
	wc2.AssertChange()
	wc2.AssertNoChange()

	// add action on unit2 and makes sure unit1 watcher doesn't trigger
	// and unit2 watcher does
	fa3, err := unit2.AddAction("snapshot", nil)
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertNoChange()
	expect2 := expectActionIds(fa3)
	wc2.AssertChange(expect2...)
	wc2.AssertNoChange()

	// add a couple actions on unit1 and make sure watcher sees events
	fa4, err := unit1.AddAction("snapshot", nil)
	c.Assert(err, jc.ErrorIsNil)
	fa5, err := unit1.AddAction("snapshot", nil)
	c.Assert(err, jc.ErrorIsNil)

	expect = expectActionIds(fa4, fa5)
	wc.AssertChange(expect...)
	wc.AssertNoChange()
}