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 [][]instance.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 *singularSuite) TestWithIsMasterTrue(c *gc.C) {
	// When IsMaster returns true, workers get started on the underlying
	// runner as usual.
	s.PatchValue(&singular.PingInterval, 1*time.Millisecond)
	underlyingRunner := newRunner()
	conn := &fakeConn{
		isMaster: true,
	}
	r, err := singular.New(underlyingRunner, conn)
	c.Assert(err, gc.IsNil)

	started := make(chan struct{}, 1)
	err = r.StartWorker("worker", func() (worker.Worker, error) {
		return worker.NewSimpleWorker(func(stop <-chan struct{}) error {
			started <- struct{}{}
			<-stop
			return nil
		}), nil
	})
	select {
	case <-started:
	case <-time.After(testing.LongWait):
		c.Fatalf("timed out waiting for worker to start")
	}

	err = worker.Stop(r)
	c.Assert(err, gc.IsNil)
}

func (s *workerSuite) TestStateServersArePublished(c *gc.C) {
	publishCh := make(chan [][]instance.HostPort)
	publish := func(apiServers [][]instance.HostPort, instanceIds []instance.Id) error {
		publishCh <- apiServers
		return nil
	}

	st := newFakeState()
	initState(c, st, 3)
	w := newWorker(st, publisherFunc(publish))
	defer func() {
		c.Check(worker.Stop(w), gc.IsNil)
	}()
	select {
	case servers := <-publishCh:
		assertAPIHostPorts(c, servers, expectedAPIHostPorts(3))
	case <-time.After(coretesting.LongWait):
		c.Fatalf("timed out waiting for publish")
	}

	// Change one of the servers' API addresses and check that it's published.

	newMachine10APIHostPorts := addressesWithPort(apiPort, "0.2.8.124")
	st.machine("10").setAPIHostPorts(newMachine10APIHostPorts)
	select {
	case servers := <-publishCh:
		expected := expectedAPIHostPorts(3)
		expected[0] = newMachine10APIHostPorts
		assertAPIHostPorts(c, servers, expected)
	case <-time.After(coretesting.LongWait):
		c.Fatalf("timed out waiting for publish")
	}
}

func (s *workerSuite) TestWorkerPublishesInstanceIds(c *gc.C) {
	s.PatchValue(&pollInterval, coretesting.LongWait+time.Second)
	s.PatchValue(&initialRetryInterval, 5*time.Millisecond)
	s.PatchValue(&maxRetryInterval, initialRetryInterval)

	publishCh := make(chan []instance.Id, 100)

	publish := func(apiServers [][]instance.HostPort, instanceIds []instance.Id) error {
		publishCh <- instanceIds
		return nil
	}
	st := newFakeState()
	initState(c, st, 3)

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

	select {
	case instanceIds := <-publishCh:
		c.Assert(instanceIds, jc.SameContents, []instance.Id{"id-10", "id-11", "id-12"})
	case <-time.After(coretesting.LongWait):
		c.Errorf("timed out waiting for publish")
	}
}

func (*runnerSuite) TestOneWorkerStartWhenStopping(c *gc.C) {
	worker.RestartDelay = 3 * time.Second
	runner := worker.NewRunner(allFatal, noImportance)
	starter := newTestWorkerStarter()
	starter.stopWait = make(chan struct{})

	err := runner.StartWorker("id", testWorkerStart(starter))
	c.Assert(err, gc.IsNil)
	starter.assertStarted(c, true)
	err = runner.StopWorker("id")
	c.Assert(err, gc.IsNil)
	err = runner.StartWorker("id", testWorkerStart(starter))
	c.Assert(err, gc.IsNil)

	close(starter.stopWait)
	starter.assertStarted(c, false)
	// Check that the task is restarted immediately without
	// the usual restart timeout delay.
	t0 := time.Now()
	starter.assertStarted(c, true)
	restartDuration := time.Since(t0)
	if restartDuration > 1*time.Second {
		c.Fatalf("task did not restart immediately")
	}
	c.Assert(worker.Stop(runner), gc.IsNil)
}

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

	memberWatcher := st.session.members.Watch()
	mustNext(c, memberWatcher)
	assertMembers(c, memberWatcher.Value(), mkMembers("0v"))

	logger.Infof("starting worker")
	w := newWorker(st, noPublisher{})
	defer func() {
		c.Check(worker.Stop(w), gc.IsNil)
	}()

	// Wait for the worker to set the initial members.
	mustNext(c, memberWatcher)
	assertMembers(c, memberWatcher.Value(), mkMembers("0v 1 2"))

	// Change an address and wait for it to be changed in the
	// members.
	st.machine("11").setStateHostPort("0.1.99.99:9876")

	mustNext(c, memberWatcher)
	expectMembers := mkMembers("0v 1 2")
	expectMembers[1].Address = "0.1.99.99:9876"
	assertMembers(c, memberWatcher.Value(), expectMembers)
}

func (s *notifyWorkerSuite) TestStop(c *gc.C) {
	err := worker.Stop(s.worker)
	c.Assert(err, gc.IsNil)
	// After stop, Wait should return right away
	err = waitShort(c, s.worker)
	c.Assert(err, gc.IsNil)
}

func (s *MachinerSuite) TestSetDead(c *gc.C) {
	mr := s.makeMachiner()
	defer worker.Stop(mr)
	c.Assert(s.machine.Destroy(), gc.IsNil)
	s.State.StartSync()
	c.Assert(mr.Wait(), gc.Equals, worker.ErrTerminateAgent)
	c.Assert(s.machine.Refresh(), gc.IsNil)
	c.Assert(s.machine.Life(), gc.Equals, state.Dead)
}

func (*runnerSuite) TestOneWorkerStart(c *gc.C) {
	runner := worker.NewRunner(noneFatal, noImportance)
	starter := newTestWorkerStarter()
	err := runner.StartWorker("id", testWorkerStart(starter))
	c.Assert(err, gc.IsNil)
	starter.assertStarted(c, true)

	c.Assert(worker.Stop(runner), gc.IsNil)
	starter.assertStarted(c, false)
}

func (s *MachineEnvironmentWatcherSuite) TestInitialStateLocalMachine1(c *gc.C) {
	proxySettings, aptProxySettings := s.updateConfig(c)

	agentConfig := agentConfig("1", provider.Local)
	envWorker := s.makeWorker(c, agentConfig)
	defer worker.Stop(envWorker)

	s.waitProxySettings(c, proxySettings)
	s.waitForFile(c, s.proxyFile, proxySettings.AsScriptEnvironment()+"\n")
	s.waitForFile(c, utils.AptConfFile, utils.AptProxyContent(aptProxySettings)+"\n")
}

func (s *MachinerSuite) TestStartSetsStatus(c *gc.C) {
	status, info, _, err := s.machine.Status()
	c.Assert(err, gc.IsNil)
	c.Assert(status, gc.Equals, params.StatusPending)
	c.Assert(info, gc.Equals, "")

	mr := s.makeMachiner()
	defer worker.Stop(mr)

	s.waitMachineStatus(c, s.machine, params.StatusStarted)
}

func (s *MachineEnvironmentWatcherSuite) TestRespondsToEvents(c *gc.C) {
	agentConfig := agentConfig("0", "ec2")
	envWorker := s.makeWorker(c, agentConfig)
	defer worker.Stop(envWorker)
	s.waitForPostSetup(c)

	proxySettings, aptProxySettings := s.updateConfig(c)

	s.waitProxySettings(c, proxySettings)
	s.waitForFile(c, s.proxyFile, proxySettings.AsScriptEnvironment()+"\n")
	s.waitForFile(c, utils.AptConfFile, utils.AptProxyContent(aptProxySettings)+"\n")
}

func (s *MachineEnvironmentWatcherSuite) TestInitialStateLocalMachine0(c *gc.C) {
	proxySettings, _ := s.updateConfig(c)

	agentConfig := agentConfig("0", provider.Local)
	envWorker := s.makeWorker(c, agentConfig)
	defer worker.Stop(envWorker)
	s.waitForPostSetup(c)

	s.waitProxySettings(c, proxySettings)

	c.Assert(utils.AptConfFile, jc.DoesNotExist)
	c.Assert(s.proxyFile, jc.DoesNotExist)
}

func (s *stringsWorkerSuite) TestChangesTriggerHandler(c *gc.C) {
	s.actor.CheckActions(c, "setup")
	s.actor.watcher.TriggerChange(c, []string{"aa", "bb"})
	waitForHandledStrings(c, s.actor.handled, []string{"aa", "bb"})
	s.actor.CheckActions(c, "setup", "handler")
	s.actor.watcher.TriggerChange(c, []string{"cc", "dd"})
	waitForHandledStrings(c, s.actor.handled, []string{"cc", "dd"})
	s.actor.watcher.TriggerChange(c, []string{"ee", "ff"})
	waitForHandledStrings(c, s.actor.handled, []string{"ee", "ff"})
	s.actor.CheckActions(c, "setup", "handler", "handler", "handler")
	c.Assert(worker.Stop(s.worker), gc.IsNil)
	s.actor.CheckActions(c, "setup", "handler", "handler", "handler", "teardown")
}

func (s *notifyWorkerSuite) TestChangesTriggerHandler(c *gc.C) {
	s.actor.CheckActions(c, "setup")
	s.actor.watcher.TriggerChange(c)
	waitForHandledNotify(c, s.actor.handled)
	s.actor.CheckActions(c, "setup", "handler")
	s.actor.watcher.TriggerChange(c)
	waitForHandledNotify(c, s.actor.handled)
	s.actor.watcher.TriggerChange(c)
	waitForHandledNotify(c, s.actor.handled)
	s.actor.CheckActions(c, "setup", "handler", "handler", "handler")
	c.Assert(worker.Stop(s.worker), gc.IsNil)
	s.actor.CheckActions(c, "setup", "handler", "handler", "handler", "teardown")
}

// During teardown we try to stop the worker, but don't hang the test suite if
// Stop never returns
func (s *notifyWorkerSuite) stopWorker(c *gc.C) {
	if s.worker == nil {
		return
	}
	done := make(chan error)
	go func() {
		done <- worker.Stop(s.worker)
	}()
	err := waitForTimeout(c, done, coretesting.LongWait)
	c.Check(err, gc.IsNil)
	s.actor = nil
	s.worker = nil
}

func (u *updaterWorker) loop() (err error) {
	u.observer, err = worker.NewEnvironObserver(u.st)
	if err != nil {
		return err
	}
	u.aggregator = newAggregator(u.observer.Environ())
	logger.Infof("instance poller received inital environment configuration")
	defer func() {
		obsErr := worker.Stop(u.observer)
		if err == nil {
			err = obsErr
		}
	}()
	return watchMachinesLoop(u, u.st.WatchEnvironMachines())
}

func (s *LoggerSuite) TestInitialState(c *gc.C) {
	config, err := s.State.EnvironConfig()
	c.Assert(err, gc.IsNil)
	expected := config.LoggingConfig()

	initial := "<root>=DEBUG;wibble=ERROR"
	c.Assert(expected, gc.Not(gc.Equals), initial)

	loggo.ResetLoggers()
	err = loggo.ConfigureLoggers(initial)
	c.Assert(err, gc.IsNil)

	loggingWorker, _ := s.makeLogger(c)
	defer worker.Stop(loggingWorker)

	s.waitLoggingInfo(c, expected)
}

func (*runnerSuite) TestOneWorkerRestart(c *gc.C) {
	runner := worker.NewRunner(noneFatal, noImportance)
	starter := newTestWorkerStarter()
	err := runner.StartWorker("id", testWorkerStart(starter))
	c.Assert(err, gc.IsNil)
	starter.assertStarted(c, true)

	// Check it restarts a few times time.
	for i := 0; i < 3; i++ {
		starter.die <- fmt.Errorf("an error")
		starter.assertStarted(c, false)
		starter.assertStarted(c, true)
	}

	c.Assert(worker.Stop(runner), gc.IsNil)
	starter.assertStarted(c, false)
}

func (s *CleanerSuite) TestCleaner(c *gc.C) {
	cr := cleaner.NewCleaner(s.State)
	defer func() { c.Assert(worker.Stop(cr), gc.IsNil) }()

	needed, err := s.State.NeedsCleanup()
	c.Assert(err, gc.IsNil)
	c.Assert(needed, gc.Equals, false)

	s.AddTestingService(c, "wordpress", s.AddTestingCharm(c, "wordpress"))
	s.AddTestingService(c, "mysql", s.AddTestingCharm(c, "mysql"))
	eps, err := s.State.InferEndpoints([]string{"wordpress", "mysql"})
	c.Assert(err, gc.IsNil)
	relM, err := s.State.AddRelation(eps...)
	c.Assert(err, gc.IsNil)

	needed, err = s.State.NeedsCleanup()
	c.Assert(err, gc.IsNil)
	c.Assert(needed, gc.Equals, false)

	// Observe destroying of the relation with a watcher.
	cw := s.State.WatchCleanups()
	defer func() { c.Assert(cw.Stop(), gc.IsNil) }()

	err = relM.Destroy()
	c.Assert(err, gc.IsNil)

	timeout := time.After(coretesting.LongWait)
	for {
		s.State.StartSync()
		select {
		case <-time.After(coretesting.ShortWait):
			continue
		case <-timeout:
			c.Fatalf("timed out waiting for cleanup")
		case <-cw.Changes():
			needed, err = s.State.NeedsCleanup()
			c.Assert(err, gc.IsNil)
			if needed {
				continue
			}
		}
		break
	}
}