func TestWatch(t *testing.T) {
	client := newEtcdClient()
	helper := tools.EtcdHelper{Client: client, Codec: runtime.DefaultCodec, ResourceVersioner: runtime.DefaultResourceVersioner}
	withEtcdKey(func(key string) {
		resp, err := client.Set(key, runtime.DefaultScheme.EncodeOrDie(&api.Pod{JSONBase: api.JSONBase{ID: "foo"}}), 0)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		expectedVersion := resp.Node.ModifiedIndex

		// watch should load the object at the current index
		w, err := helper.Watch(key, 0)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		event := <-w.ResultChan()
		if event.Type != watch.Added || event.Object == nil {
			t.Fatalf("expected first value to be set to ADDED, got %#v", event)
		}

		// version should match what we set
		pod := event.Object.(*api.Pod)
		if pod.ResourceVersion != expectedVersion {
			t.Errorf("expected version %d, got %#v", expectedVersion, pod)
		}

		// should be no events in the stream
		select {
		case event, ok := <-w.ResultChan():
			if !ok {
				t.Fatalf("channel closed unexpectedly")
			}
			t.Fatalf("unexpected object in channel: %#v", event)
		default:
		}

		// should return the previously deleted item in the watch, but with the latest index
		resp, err = client.Delete(key, false)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		expectedVersion = resp.Node.ModifiedIndex
		event = <-w.ResultChan()
		if event.Type != watch.Deleted {
			t.Errorf("expected deleted event %#v", event)
		}
		pod = event.Object.(*api.Pod)
		if pod.ResourceVersion != expectedVersion {
			t.Errorf("expected version %d, got %#v", expectedVersion, pod)
		}
	})
}

func (rm *ReplicationManager) synchronize() {
	var controllerSpecs []api.ReplicationController
	helper := tools.EtcdHelper{rm.etcdClient}
	err := helper.ExtractList("/registry/controllers", &controllerSpecs)
	if err != nil {
		glog.Errorf("Synchronization error: %v (%#v)", err, err)
		return
	}
	for _, controllerSpec := range controllerSpecs {
		err = rm.syncHandler(controllerSpec)
		if err != nil {
			glog.Errorf("Error synchronizing: %#v", err)
		}
	}
}

func TestSetObj(t *testing.T) {
	client := newEtcdClient()
	helper := tools.EtcdHelper{Client: client, Codec: stringCodec{}}
	withEtcdKey(func(key string) {
		if err := helper.SetObj(key, "object"); err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		resp, err := client.Get(key, false, false)
		if err != nil || resp.Node == nil {
			t.Fatalf("unexpected error: %v %v", err, resp)
		}
		if resp.Node.Value != "object" {
			t.Errorf("unexpected response: %#v", resp.Node)
		}
	})
}

func TestExtractObj(t *testing.T) {
	client := framework.NewEtcdClient()
	helper := tools.EtcdHelper{Client: client, Codec: stringCodec{}}
	framework.WithEtcdKey(func(key string) {
		_, err := client.Set(key, "object", 0)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		s := fakeAPIObject("")
		if err := helper.ExtractObj(key, &s, false); err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		if s != "object" {
			t.Errorf("unexpected response: %#v", s)
		}
	})
}

func (rm *ReplicationManager) synchronize() {
	var controllerSpecs []api.ReplicationController
	helper := tools.EtcdHelper{rm.etcdClient}
	err := helper.ExtractList("/registry/controllers", &controllerSpecs)
	if err != nil {
		glog.Errorf("Synchronization error: %v (%#v)", err, err)
		return
	}
	wg := sync.WaitGroup{}
	wg.Add(len(controllerSpecs))
	for ix := range controllerSpecs {
		go func(ix int) {
			defer wg.Done()
			err := rm.syncHandler(controllerSpecs[ix])
			if err != nil {
				glog.Errorf("Error synchronizing: %#v", err)
			}
		}(ix)
	}
	wg.Wait()
}

// makeEtcdWatch starts watching via etcd.
func (rm *ReplicationManager) makeEtcdWatch() (watch.Interface, error) {
	helper := tools.EtcdHelper{rm.etcdClient}
	return helper.WatchList("/registry/controllers", tools.Everything)
}