func newStorage(t *testing.T) (*REST, *StatusREST, *tools.FakeEtcdClient, storage.Interface) {
	fakeEtcdClient := tools.NewFakeEtcdClient(t)
	fakeEtcdClient.TestIndex = true
	etcdStorage := etcdstorage.NewEtcdStorage(fakeEtcdClient, latest.Codec, etcdtest.PathPrefix())
	storage, statusStorage := NewStorage(etcdStorage)
	return storage, statusStorage, fakeEtcdClient, etcdStorage
}

// startMasterOrDie starts a kubernetes master and an httpserver to handle api requests
func startMasterOrDie(masterConfig *master.Config) (*master.Master, *httptest.Server, *tools.EtcdHelper) {
	var m *master.Master
	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		m.Handler.ServeHTTP(w, req)
	}))

	var helper tools.EtcdHelper
	var err error
	if masterConfig == nil {
		helper, err = master.NewEtcdHelper(NewEtcdClient(), "", etcdtest.PathPrefix())
		if err != nil {
			glog.Fatalf("Failed to create etcd helper for master %v", err)
		}
		masterConfig = &master.Config{
			EtcdHelper:        helper,
			KubeletClient:     client.FakeKubeletClient{},
			EnableLogsSupport: false,
			EnableProfiling:   true,
			EnableUISupport:   false,
			APIPrefix:         "/api",
			Authorizer:        apiserver.NewAlwaysAllowAuthorizer(),
			AdmissionControl:  admit.NewAlwaysAdmit(),
		}
	} else {
		helper = masterConfig.EtcdHelper
	}
	m = master.New(masterConfig)
	return m, s, &helper
}

func NewTestEventEtcdRegistry(t *testing.T) (*tools.FakeEtcdClient, generic.Registry) {
	f := tools.NewFakeEtcdClient(t)
	f.TestIndex = true

	h := tools.NewEtcdHelper(f, testapi.Codec(), etcdtest.PathPrefix())
	return f, NewEtcdRegistry(h, testTTL)
}

func TestSetObjWithoutResourceVersioner(t *testing.T) {
	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
	fakeClient := NewFakeEtcdClient(t)
	helper := NewEtcdHelper(fakeClient, testapi.Codec(), etcdtest.PathPrefix())
	helper.Versioner = nil
	returnedObj := &api.Pod{}
	err := helper.SetObj("/some/key", obj, returnedObj, 3)
	key := etcdtest.AddPrefix("/some/key")
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
	data, err := testapi.Codec().Encode(obj)
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
	expect := string(data)
	got := fakeClient.Data[key].R.Node.Value
	if expect != got {
		t.Errorf("Wanted %v, got %v", expect, got)
	}
	if e, a := uint64(3), fakeClient.LastSetTTL; e != a {
		t.Errorf("Wanted %v, got %v", e, a)
	}
	if obj.ResourceVersion != returnedObj.ResourceVersion || obj.Name != returnedObj.Name {
		t.Errorf("If set was successful but returned object did not have correct resource version")
	}
}

func TestWatchPurposefulShutdown(t *testing.T) {
	fakeClient := tools.NewFakeEtcdClient(t)

	h := newEtcdHelper(fakeClient, codec, etcdtest.PathPrefix())
	key := "/some/key"
	prefixedKey := etcdtest.AddPrefix(key)
	fakeClient.ExpectNotFoundGet(prefixedKey)

	// Test purposeful shutdown
	watching, err := h.Watch(key, 0, storage.Everything)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	fakeClient.WaitForWatchCompletion()
	watching.Stop()

	// Did everything shut down?
	if _, open := <-fakeClient.WatchResponse; open {
		t.Errorf("A stop did not cause a graceful shutdown")
	}
	if _, open := <-watching.ResultChan(); open {
		t.Errorf("An injected error did not cause a graceful shutdown")
	}
}

func TestGuaranteedUpdateNoChange(t *testing.T) {
	fakeClient := NewFakeEtcdClient(t)
	fakeClient.TestIndex = true
	helper := NewEtcdHelper(fakeClient, codec, etcdtest.PathPrefix())
	key := etcdtest.AddPrefix("/some/key")

	// Create a new node.
	fakeClient.ExpectNotFoundGet(key)
	obj := &TestResource{ObjectMeta: api.ObjectMeta{Name: "foo"}, Value: 1}
	err := helper.GuaranteedUpdate("/some/key", &TestResource{}, true, SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
		return obj, nil
	}))
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}

	// Update an existing node with the same data
	callbackCalled := false
	objUpdate := &TestResource{ObjectMeta: api.ObjectMeta{Name: "foo"}, Value: 1}
	err = helper.GuaranteedUpdate("/some/key", &TestResource{}, true, SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
		fakeClient.Err = errors.New("should not be called")
		callbackCalled = true
		return objUpdate, nil
	}))
	if err != nil {
		t.Fatalf("Unexpected error %#v", err)
	}
	if !callbackCalled {
		t.Errorf("tryUpdate callback should have been called.")
	}
}

func TestWatchFromNotFound(t *testing.T) {
	fakeClient := tools.NewFakeEtcdClient(t)
	key := "/some/key"
	prefixedKey := etcdtest.AddPrefix(key)
	fakeClient.Data[prefixedKey] = tools.EtcdResponseWithError{
		R: &etcd.Response{
			Node: nil,
		},
		E: &etcd.EtcdError{
			Index:     2,
			ErrorCode: 100,
		},
	}
	h := newEtcdHelper(fakeClient, codec, etcdtest.PathPrefix())

	watching, err := h.Watch(key, 0, storage.Everything)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	fakeClient.WaitForWatchCompletion()
	if fakeClient.WatchIndex != 3 {
		t.Errorf("Expected client to wait for %d, got %#v", 3, fakeClient)
	}

	watching.Stop()
}

func TestWatchEtcdError(t *testing.T) {
	codec := latest.Codec
	fakeClient := tools.NewFakeEtcdClient(t)
	fakeClient.ExpectNotFoundGet("/some/key")
	fakeClient.WatchImmediateError = fmt.Errorf("immediate error")
	h := newEtcdHelper(fakeClient, codec, etcdtest.PathPrefix())

	watching, err := h.Watch("/some/key", 4, storage.Everything)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer watching.Stop()

	got := <-watching.ResultChan()
	if got.Type != watch.Error {
		t.Fatalf("Unexpected non-error")
	}
	status, ok := got.Object.(*api.Status)
	if !ok {
		t.Fatalf("Unexpected non-error object type")
	}
	if status.Message != "immediate error" {
		t.Errorf("Unexpected wrong error")
	}
	if status.Status != api.StatusFailure {
		t.Errorf("Unexpected wrong error status")
	}
}

func TestGuaranteedUpdateKeyNotFound(t *testing.T) {
	fakeClient := NewFakeEtcdClient(t)
	fakeClient.TestIndex = true
	helper := NewEtcdHelper(fakeClient, codec, etcdtest.PathPrefix())
	key := etcdtest.AddPrefix("/some/key")

	// Create a new node.
	fakeClient.ExpectNotFoundGet(key)
	obj := &TestResource{ObjectMeta: api.ObjectMeta{Name: "foo"}, Value: 1}

	f := SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
		return obj, nil
	})

	ignoreNotFound := false
	err := helper.GuaranteedUpdate("/some/key", &TestResource{}, ignoreNotFound, f)
	if err == nil {
		t.Errorf("Expected error for key not found.")
	}

	ignoreNotFound = true
	err = helper.GuaranteedUpdate("/some/key", &TestResource{}, ignoreNotFound, f)
	if err != nil {
		t.Errorf("Unexpected error %v.", err)
	}
}

func NewTestEtcdRegistryWithPods(client tools.EtcdClient) *Registry {
	helper := tools.NewEtcdHelper(client, latest.Codec, etcdtest.PathPrefix())
	podStorage := podetcd.NewStorage(helper, nil)
	endpointStorage := endpointetcd.NewStorage(helper)
	registry := NewRegistry(helper, pod.NewRegistry(podStorage.Pod), endpoint.NewRegistry(endpointStorage))
	return registry
}

func TestGetServersToValidate(t *testing.T) {
	master := Master{}
	config := Config{}
	fakeClient := tools.NewFakeEtcdClient(t)
	fakeClient.Machines = []string{"http://machine1:4001", "http://machine2", "http://machine3:4003"}
	config.EtcdHelper = tools.NewEtcdHelper(fakeClient, latest.Codec, etcdtest.PathPrefix())
	config.EtcdHelper.Versioner = nil

	master.nodeRegistry = registrytest.NewMinionRegistry([]string{"node1", "node2"}, api.NodeResources{})

	servers := master.getServersToValidate(&config, true)

	if len(servers) != 7 {
		t.Errorf("unexpected server list: %#v", servers)
	}
	for _, server := range []string{"scheduler", "controller-manager", "etcd-0", "etcd-1", "etcd-2", "node-0", "node-1"} {
		if _, ok := servers[server]; !ok {
			t.Errorf("server list missing: %s", server)
		}
	}

	servers = master.getServersToValidate(&config, false)

	if len(servers) != 5 {
		t.Errorf("unexpected server list: %#v", servers)
	}
	for _, server := range []string{"scheduler", "controller-manager", "etcd-0", "etcd-1", "etcd-2"} {
		if _, ok := servers[server]; !ok {
			t.Errorf("server list missing: %s", server)
		}
	}
}

func runAMaster(t *testing.T) (*master.Master, *httptest.Server) {
	helper, err := master.NewEtcdHelper(newEtcdClient(), testapi.Version(), etcdtest.PathPrefix())
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}

	m := master.New(&master.Config{
		EtcdHelper:            helper,
		KubeletClient:         client.FakeKubeletClient{},
		EnableCoreControllers: true,
		EnableLogsSupport:     false,
		EnableProfiling:       true,
		EnableUISupport:       false,
		APIPrefix:             "/api",
		// Enable v1beta3 if we are testing that version.
		EnableV1Beta3:    testapi.Version() == "v1beta3",
		Authorizer:       apiserver.NewAlwaysAllowAuthorizer(),
		AdmissionControl: admit.NewAlwaysAdmit(),
	})

	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		m.Handler.ServeHTTP(w, req)
	}))

	return m, s
}

func newStorage(t *testing.T) (*REST, *StatusREST, *tools.FakeEtcdClient, tools.EtcdHelper) {
	fakeEtcdClient := tools.NewFakeEtcdClient(t)
	fakeEtcdClient.TestIndex = true
	helper := tools.NewEtcdHelper(fakeEtcdClient, latest.Codec, etcdtest.PathPrefix())
	storage, statusStorage := NewStorage(helper)
	return storage, statusStorage, fakeEtcdClient, helper
}

func TestWatch(t *testing.T) {
	codec := latest.Codec
	fakeClient := tools.NewFakeEtcdClient(t)
	key := "/some/key"
	prefixedKey := etcdtest.AddPrefix(key)
	fakeClient.ExpectNotFoundGet(prefixedKey)
	h := newEtcdHelper(fakeClient, codec, etcdtest.PathPrefix())

	watching, err := h.Watch(key, 0, storage.Everything)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	fakeClient.WaitForWatchCompletion()
	// when server returns not found, the watch index starts at the next value (1)
	if fakeClient.WatchIndex != 1 {
		t.Errorf("Expected client to be at index %d, got %#v", 1, fakeClient)
	}

	// Test normal case
	pod := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
	podBytes, _ := codec.Encode(pod)
	fakeClient.WatchResponse <- &etcd.Response{
		Action: "set",
		Node: &etcd.Node{
			Value: string(podBytes),
		},
	}

	event := <-watching.ResultChan()
	if e, a := watch.Added, event.Type; e != a {
		t.Errorf("Expected %v, got %v", e, a)
	}
	if e, a := pod, event.Object; !api.Semantic.DeepDerivative(e, a) {
		t.Errorf("Expected %v, got %v", e, a)
	}

	// Test error case
	fakeClient.WatchInjectError <- fmt.Errorf("Injected error")

	if errEvent, ok := <-watching.ResultChan(); !ok {
		t.Errorf("no error result?")
	} else {
		if e, a := watch.Error, errEvent.Type; e != a {
			t.Errorf("Expected %v, got %v", e, a)
		}
		if e, a := "Injected error", errEvent.Object.(*api.Status).Message; e != a {
			t.Errorf("Expected %v, got %v", e, a)
		}
	}

	// Did everything shut down?
	if _, open := <-fakeClient.WatchResponse; open {
		t.Errorf("An injected error did not cause a graceful shutdown")
	}
	if _, open := <-watching.ResultChan(); open {
		t.Errorf("An injected error did not cause a graceful shutdown")
	}
}

func TestCreateObjNilOutParam(t *testing.T) {
	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
	fakeClient := NewFakeEtcdClient(t)
	helper := NewEtcdHelper(fakeClient, testapi.Codec(), etcdtest.PathPrefix())
	err := helper.CreateObj("/some/key", obj, nil, 5)
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
}

func TestSetNilOutParam(t *testing.T) {
	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
	fakeClient := tools.NewFakeEtcdClient(t)
	helper := newEtcdHelper(fakeClient, testapi.Codec(), etcdtest.PathPrefix())
	helper.versioner = nil
	err := helper.Set("/some/key", obj, nil, 3)
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
}

func TestSetObjFailCAS(t *testing.T) {
	obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "1"}}
	fakeClient := NewFakeEtcdClient(t)
	fakeClient.CasErr = fakeClient.NewError(123)
	helper := NewEtcdHelper(fakeClient, testapi.Codec(), etcdtest.PathPrefix())
	err := helper.SetObj("/some/key", obj, nil, 5)
	if err == nil {
		t.Errorf("Expecting error.")
	}
}

func setup(t *testing.T) (*tools.FakeEtcdClient, tools.EtcdHelper, *REST) {
	fakeEtcdClient := tools.NewFakeEtcdClient(t)
	fakeEtcdClient.TestIndex = true
	helper := tools.NewEtcdHelper(fakeEtcdClient, latest.Codec, etcdtest.PathPrefix())
	imageStorage := imageetcd.NewREST(helper)
	imageRegistry := image.NewRegistry(imageStorage)
	imageStreamStorage, imageStreamStatus := imagestreametcd.NewREST(helper, testDefaultRegistry, &fakeSubjectAccessReviewRegistry{})
	imageStreamRegistry := imagestream.NewRegistry(imageStreamStorage, imageStreamStatus)
	storage := NewREST(imageRegistry, imageStreamRegistry)
	return fakeEtcdClient, helper, storage
}

func TestGuaranteedUpdate(t *testing.T) {
	fakeClient := NewFakeEtcdClient(t)
	fakeClient.TestIndex = true
	helper := NewEtcdHelper(fakeClient, codec, etcdtest.PathPrefix())
	key := etcdtest.AddPrefix("/some/key")

	// Create a new node.
	fakeClient.ExpectNotFoundGet(key)
	obj := &TestResource{ObjectMeta: api.ObjectMeta{Name: "foo"}, Value: 1}
	err := helper.GuaranteedUpdate("/some/key", &TestResource{}, true, SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
		return obj, nil
	}))
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
	data, err := codec.Encode(obj)
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
	expect := string(data)
	got := fakeClient.Data[key].R.Node.Value
	if expect != got {
		t.Errorf("Wanted %v, got %v", expect, got)
	}

	// Update an existing node.
	callbackCalled := false
	objUpdate := &TestResource{ObjectMeta: api.ObjectMeta{Name: "foo"}, Value: 2}
	err = helper.GuaranteedUpdate("/some/key", &TestResource{}, true, SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
		callbackCalled = true

		if in.(*TestResource).Value != 1 {
			t.Errorf("Callback input was not current set value")
		}

		return objUpdate, nil
	}))
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
	data, err = codec.Encode(objUpdate)
	if err != nil {
		t.Errorf("Unexpected error %#v", err)
	}
	expect = string(data)
	got = fakeClient.Data[key].R.Node.Value
	if expect != got {
		t.Errorf("Wanted %v, got %v", expect, got)
	}

	if !callbackCalled {
		t.Errorf("tryUpdate callback should have been called.")
	}
}

func TestWatch(t *testing.T) {
	client := framework.NewEtcdClient()
	etcdStorage := tools.NewEtcdStorage(client, testapi.Codec(), etcdtest.PathPrefix())
	framework.WithEtcdKey(func(key string) {
		key = etcdtest.AddPrefix(key)
		resp, err := client.Set(key, runtime.EncodeOrDie(testapi.Codec(), &api.Pod{ObjectMeta: api.ObjectMeta{Name: "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 := etcdStorage.Watch(key, 0, tools.Everything)
		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 != strconv.FormatUint(expectedVersion, 10) {
			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 != strconv.FormatUint(expectedVersion, 10) {
			t.Errorf("expected version %d, got %#v", expectedVersion, pod)
		}
	})
}