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 *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 *RebootSuite) SetUpTest(c *gc.C) {
	s.ConnSuite.SetUpTest(c)
	var err error

	// Add machine
	s.machine, err = s.State.AddMachine("quantal", state.JobManageModel)
	c.Assert(err, jc.ErrorIsNil)
	// Add first container
	s.c1, err = s.State.AddMachineInsideMachine(state.MachineTemplate{
		Series: "quantal",
		Jobs:   []state.MachineJob{state.JobHostUnits},
	}, s.machine.Id(), instance.LXC)
	c.Assert(err, jc.ErrorIsNil)
	// Add second container
	s.c2, err = s.State.AddMachineInsideMachine(state.MachineTemplate{
		Series: "quantal",
		Jobs:   []state.MachineJob{state.JobHostUnits},
	}, s.c1.Id(), instance.LXC)
	c.Assert(err, jc.ErrorIsNil)

	// Add container on the same level as the first container.
	s.c3, err = s.State.AddMachineInsideMachine(state.MachineTemplate{
		Series: "quantal",
		Jobs:   []state.MachineJob{state.JobHostUnits},
	}, s.machine.Id(), instance.LXC)
	c.Assert(err, jc.ErrorIsNil)

	s.w, err = s.machine.WatchForRebootEvent()
	c.Assert(err, jc.ErrorIsNil)

	s.wc = statetesting.NewNotifyWatcherC(c, s.State, s.w)
	s.wc.AssertOneChange()

	s.wC1, err = s.c1.WatchForRebootEvent()
	c.Assert(err, jc.ErrorIsNil)

	// Initial event on container 1.
	s.wcC1 = statetesting.NewNotifyWatcherC(c, s.State, s.wC1)
	s.wcC1.AssertOneChange()

	// Get reboot watcher on container 2
	s.wC2, err = s.c2.WatchForRebootEvent()
	c.Assert(err, jc.ErrorIsNil)

	// Initial event on container 2.
	s.wcC2 = statetesting.NewNotifyWatcherC(c, s.State, s.wC2)
	s.wcC2.AssertOneChange()

	// Get reboot watcher on container 3
	s.wC3, err = s.c3.WatchForRebootEvent()
	c.Assert(err, jc.ErrorIsNil)

	// Initial event on container 3.
	s.wcC3 = statetesting.NewNotifyWatcherC(c, s.State, s.wC3)
	s.wcC3.AssertOneChange()
}

func (s *uniterSuite) TestWatchConfigSettings(c *gc.C) {
	err := s.wordpressUnit.SetCharmURL(s.wpCharm.URL())
	c.Assert(err, gc.IsNil)

	c.Assert(s.resources.Count(), gc.Equals, 0)

	args := params.Entities{Entities: []params.Entity{
		{Tag: "unit-mysql-0"},
		{Tag: "unit-wordpress-0"},
		{Tag: "unit-foo-42"},
	}}
	result, err := s.uniter.WatchConfigSettings(args)
	c.Assert(err, gc.IsNil)
	c.Assert(result, gc.DeepEquals, params.NotifyWatchResults{
		Results: []params.NotifyWatchResult{
			{Error: apiservertesting.ErrUnauthorized},
			{NotifyWatcherId: "1"},
			{Error: apiservertesting.ErrUnauthorized},
		},
	})

	// Verify the resource was registered and stop when done
	c.Assert(s.resources.Count(), gc.Equals, 1)
	resource := s.resources.Get("1")
	defer statetesting.AssertStop(c, resource)

	// Check that the Watch has consumed the initial event ("returned" in
	// the Watch call)
	wc := statetesting.NewNotifyWatcherC(c, s.State, resource.(state.NotifyWatcher))
	wc.AssertNoChange()
}

func (s *retryStrategySuite) TestWatchRetryStrategy(c *gc.C) {
	c.Assert(s.resources.Count(), gc.Equals, 0)

	args := params.Entities{Entities: []params.Entity{
		{Tag: s.unit.UnitTag().String()},
		{Tag: "unit-foo-42"},
	}}
	r, err := s.strategy.WatchRetryStrategy(args)
	c.Assert(err, jc.ErrorIsNil)
	c.Assert(r, gc.DeepEquals, params.NotifyWatchResults{
		Results: []params.NotifyWatchResult{
			{NotifyWatcherId: "1"},
			{Error: apiservertesting.ErrUnauthorized},
		},
	})

	c.Assert(s.resources.Count(), gc.Equals, 1)
	resource := s.resources.Get("1")
	defer statetesting.AssertStop(c, resource)

	wc := statetesting.NewNotifyWatcherC(c, s.State, resource.(state.NotifyWatcher))
	wc.AssertNoChange()

	s.setRetryStrategy(c, false)
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertOneChange()
}

func (s *CloudCredentialsSuite) createCredentialWatcher(c *gc.C, st *state.State, cred names.CloudCredentialTag) (
	state.NotifyWatcher, statetesting.NotifyWatcherC,
) {
	w := st.WatchCredential(cred)
	s.AddCleanup(func(c *gc.C) { statetesting.AssertStop(c, w) })
	return w, statetesting.NewNotifyWatcherC(c, st, w)
}

func (s *workerJujuConnSuite) TestPublisherSetsAPIHostPorts(c *gc.C) {
	st := newFakeState()
	initState(c, st, 3)

	watcher := s.State.WatchAPIHostPorts()
	cwatch := statetesting.NewNotifyWatcherC(c, s.State, watcher)
	cwatch.AssertOneChange()

	statePublish := newPublisher(s.State)

	// Wrap the publisher so that we can call StartSync immediately
	// after the publishAPIServers method is called.
	publish := func(apiServers [][]network.HostPort, instanceIds []instance.Id) error {
		err := statePublish.publishAPIServers(apiServers, instanceIds)
		s.State.StartSync()
		return err
	}

	w := newWorker(st, publisherFunc(publish))
	defer func() {
		c.Check(worker.Stop(w), gc.IsNil)
	}()

	cwatch.AssertOneChange()
	hps, err := s.State.APIHostPorts()
	c.Assert(err, gc.IsNil)
	assertAPIHostPorts(c, hps, expectedAPIHostPorts(3))
}

func (s *machinerSuite) TestWatch(c *gc.C) {
	c.Assert(s.resources.Count(), gc.Equals, 0)

	args := params.Entities{Entities: []params.Entity{
		{Tag: "machine-1"},
		{Tag: "machine-0"},
		{Tag: "machine-42"},
	}}
	result, err := s.machiner.Watch(args)
	c.Assert(err, gc.IsNil)
	c.Assert(result, gc.DeepEquals, params.NotifyWatchResults{
		Results: []params.NotifyWatchResult{
			{NotifyWatcherId: "1"},
			{Error: apiservertesting.ErrUnauthorized},
			{Error: apiservertesting.ErrUnauthorized},
		},
	})

	// Verify the resource was registered and stop when done
	c.Assert(s.resources.Count(), gc.Equals, 1)
	c.Assert(result.Results[0].NotifyWatcherId, gc.Equals, "1")
	resource := s.resources.Get("1")
	defer statetesting.AssertStop(c, resource)

	// Check that the Watch has consumed the initial event ("returned" in
	// the Watch call)
	wc := statetesting.NewNotifyWatcherC(c, s.State, resource.(state.NotifyWatcher))
	wc.AssertNoChange()
}

func (s *VolumeStateSuite) TestWatchVolumeAttachment(c *gc.C) {
	_, u, storageTag := s.setupSingleStorage(c, "block", "loop-pool")
	err := s.State.AssignUnit(u, state.AssignCleanEmpty)
	c.Assert(err, jc.ErrorIsNil)
	assignedMachineId, err := u.AssignedMachineId()
	c.Assert(err, jc.ErrorIsNil)
	machineTag := names.NewMachineTag(assignedMachineId)

	volume := s.storageInstanceVolume(c, storageTag)
	volumeTag := volume.VolumeTag()

	w := s.State.WatchVolumeAttachment(machineTag, volumeTag)
	defer testing.AssertStop(c, w)
	wc := testing.NewNotifyWatcherC(c, s.State, w)
	wc.AssertOneChange()

	machine, err := s.State.Machine(assignedMachineId)
	c.Assert(err, jc.ErrorIsNil)
	err = machine.SetProvisioned("inst-id", "fake_nonce", nil)
	c.Assert(err, jc.ErrorIsNil)

	// volume attachment will NOT react to volume changes
	err = s.State.SetVolumeInfo(volumeTag, state.VolumeInfo{VolumeId: "vol-123"})
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertNoChange()

	err = s.State.SetVolumeAttachmentInfo(
		machineTag, volumeTag, state.VolumeAttachmentInfo{
			DeviceName: "xvdf1",
		},
	)
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertOneChange()
}

// TestWatchMeterStatus tests the meter status watcher functionality.
func TestWatchMeterStatus(c *gc.C, status meterstatus.MeterStatus, unit *jujustate.Unit, state *jujustate.State, resources *common.Resources) {
	c.Assert(resources.Count(), gc.Equals, 0)

	args := params.Entities{Entities: []params.Entity{
		{Tag: unit.UnitTag().String()},
		{Tag: "unit-foo-42"},
	}}
	result, err := status.WatchMeterStatus(args)
	c.Assert(err, jc.ErrorIsNil)
	c.Assert(result, gc.DeepEquals, params.NotifyWatchResults{
		Results: []params.NotifyWatchResult{
			{NotifyWatcherId: "1"},
			{Error: apiservertesting.ErrUnauthorized},
		},
	})

	// Verify the resource was registered and stop when done
	c.Assert(resources.Count(), gc.Equals, 1)
	resource := resources.Get("1")
	defer statetesting.AssertStop(c, resource)

	// Check that the Watch has consumed the initial event ("returned" in
	// the Watch call)
	wc := statetesting.NewNotifyWatcherC(c, state, resource.(jujustate.NotifyWatcher))
	wc.AssertNoChange()

	err = unit.SetMeterStatus("GREEN", "No additional information.")
	c.Assert(err, jc.ErrorIsNil)
	wc.AssertOneChange()
}

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 *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 *ServiceLeaderSuite) TestWatchInitialEvent(c *gc.C) {
	w := s.service.WatchLeaderSettings()
	defer testing.AssertStop(c, w)

	wc := testing.NewNotifyWatcherC(c, s.State, w)
	wc.AssertOneChange()
}

func (s *authorisedKeysSuite) TestWatchAuthorisedKeys(c *gc.C) {
	args := params.Entities{
		Entities: []params.Entity{
			{Tag: s.rawMachine.Tag().String()},
			{Tag: s.unrelatedMachine.Tag().String()},
			{Tag: "machine-42"},
		},
	}
	results, err := s.keyupdater.WatchAuthorisedKeys(args)
	c.Assert(err, jc.ErrorIsNil)
	c.Assert(results, gc.DeepEquals, params.NotifyWatchResults{
		Results: []params.NotifyWatchResult{
			{NotifyWatcherId: "1"},
			{Error: apiservertesting.ErrUnauthorized},
			{Error: apiservertesting.ErrUnauthorized},
		},
	})
	c.Assert(results.Results[0].NotifyWatcherId, gc.Not(gc.Equals), "")
	c.Assert(results.Results[0].Error, gc.IsNil)
	resource := s.resources.Get(results.Results[0].NotifyWatcherId)
	c.Assert(resource, gc.NotNil)

	w := resource.(state.NotifyWatcher)
	wc := statetesting.NewNotifyWatcherC(c, s.State, w)
	wc.AssertNoChange()

	s.setAuthorizedKeys(c, "key1\nkey2")

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

func (s *networkerSuite) TestWatchInterfaces(c *gc.C) {
	c.Assert(s.resources.Count(), gc.Equals, 0)

	args := params.Entities{Entities: []params.Entity{
		{Tag: "machine-0"},
		{Tag: "machine-0-lxc-0"},
		{Tag: "machine-0-lxc-0-lxc-0"},
	}}
	result, err := s.networker.WatchInterfaces(args)
	c.Assert(err, gc.IsNil)
	c.Assert(result, gc.DeepEquals, params.NotifyWatchResults{
		Results: []params.NotifyWatchResult{
			{NotifyWatcherId: "1"},
			{NotifyWatcherId: "2"},
			{NotifyWatcherId: "3"},
		},
	})

	// Verify the resource was registered and stop when done
	c.Assert(s.resources.Count(), gc.Equals, 3)
	for _, watcherId := range []string{"1", "2", "3"} {
		resource := s.resources.Get(watcherId)
		defer statetesting.AssertStop(c, resource)

		// Check that the WatchInterfaces has consumed the initial event ("returned" in
		// the Watch call)
		wc := statetesting.NewNotifyWatcherC(c, s.State, resource.(state.NotifyWatcher))
		wc.AssertNoChange()
	}
}

func (s *MigrationSuite) createStatusWatcher(c *gc.C, st *state.State) (
	state.NotifyWatcher, statetesting.NotifyWatcherC,
) {
	w := st.WatchMigrationStatus()
	s.AddCleanup(func(c *gc.C) { statetesting.AssertStop(c, w) })
	return w, statetesting.NewNotifyWatcherC(c, st, w)
}

func (s *MachineRemovalSuite) createRemovalWatcher(c *gc.C, st *state.State) (
	state.NotifyWatcher, testing.NotifyWatcherC,
) {
	w := st.WatchMachineRemovals()
	s.AddCleanup(func(c *gc.C) { workertest.CleanKill(c, w) })
	return w, testing.NewNotifyWatcherC(c, st, w)
}

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 *withoutStateServerSuite) TestWatchMachineErrorRetry(c *gc.C) {
	s.PatchValue(&provisioner.ErrorRetryWaitDelay, 2*coretesting.ShortWait)
	c.Assert(s.resources.Count(), gc.Equals, 0)

	_, err := s.provisioner.WatchMachineErrorRetry()
	c.Assert(err, gc.IsNil)

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

	// Check that the Watch has consumed the initial event ("returned"
	// in the Watch call)
	wc := statetesting.NewNotifyWatcherC(c, s.State, resource.(state.NotifyWatcher))
	wc.AssertNoChange()

	// We should now get a time triggered change.
	wc.AssertOneChange()

	// Make sure WatchMachineErrorRetry fails with a machine agent login.
	anAuthorizer := s.authorizer
	anAuthorizer.MachineAgent = true
	anAuthorizer.EnvironManager = false
	aProvisioner, err := provisioner.NewProvisionerAPI(s.State, s.resources, anAuthorizer)
	c.Assert(err, gc.IsNil)

	result, err := aProvisioner.WatchMachineErrorRetry()
	c.Assert(err, gc.ErrorMatches, "permission denied")
	c.Assert(result, gc.DeepEquals, params.NotifyWatchResult{})
}

func (s *ModelMigrationSuite) createStatusWatcher(c *gc.C, st *state.State) (
	state.NotifyWatcher, statetesting.NotifyWatcherC,
) {
	w, err := st.WatchMigrationStatus()
	c.Assert(err, jc.ErrorIsNil)
	s.AddCleanup(func(c *gc.C) { statetesting.AssertStop(c, w) })
	return w, statetesting.NewNotifyWatcherC(c, st, w)
}