Removal of fakeClient from registry/generic/etcd/etcd_test.go in leiu of

NewEtcdTestClientServer

pkg/registry/generic/etcd/etcd.go

@@ -54,6 +54,8 @@ import (
 // logic specific to the API.
 //
 // TODO: make the default exposed methods exactly match a generic RESTStorage
+// TODO: because all aspects of etcd have been removed it should really
+//       just be called a registry implementation.
 type Etcd struct {
 	// Called to make a new object, should return e.g., &api.Pod{}
 	NewFunc func() runtime.Object

pkg/registry/generic/etcd/etcd_test.go

@@ -19,6 +19,7 @@ package etcd
 import (
 	"fmt"
 	"path"
+	"reflect"
 	"testing"
 
 	"k8s.io/kubernetes/pkg/api"
@@ -27,13 +28,11 @@ import (
 	"k8s.io/kubernetes/pkg/registry/generic"
 	"k8s.io/kubernetes/pkg/runtime"
 	etcdstorage "k8s.io/kubernetes/pkg/storage/etcd"
-	"k8s.io/kubernetes/pkg/tools"
+	etcdtesting "k8s.io/kubernetes/pkg/storage/etcd/testing"
+	storagetesting "k8s.io/kubernetes/pkg/storage/testing"
 	"k8s.io/kubernetes/pkg/tools/etcdtest"
-	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/fielderrors"
 	"k8s.io/kubernetes/pkg/util/sets"
-
-	"github.com/coreos/go-etcd/etcd"
 )
 
 type testRESTStrategy struct {
@@ -68,13 +67,13 @@ func hasCreated(t *testing.T, pod *api.Pod) func(runtime.Object) bool {
 	}
 }
 
-func NewTestGenericEtcdRegistry(t *testing.T) (*tools.FakeEtcdClient, *Etcd) {
-	f := tools.NewFakeEtcdClient(t)
-	f.TestIndex = true
-	s := etcdstorage.NewEtcdStorage(f, testapi.Default.Codec(), etcdtest.PathPrefix())
-	strategy := &testRESTStrategy{api.Scheme, api.SimpleNameGenerator, true, false, true}
+func NewTestGenericEtcdRegistry(t *testing.T) (*etcdtesting.EtcdTestServer, *Etcd) {
 	podPrefix := "/pods"
-	return f, &Etcd{
+	server := etcdtesting.NewEtcdTestClientServer(t)
+	s := etcdstorage.NewEtcdStorage(server.Client, testapi.Default.Codec(), etcdtest.PathPrefix())
+	strategy := &testRESTStrategy{api.Scheme, api.SimpleNameGenerator, true, false, true}
+
+	return server, &Etcd{
 		NewFunc:        func() runtime.Object { return &api.Pod{} },
 		NewListFunc:    func() runtime.Object { return &api.PodList{} },
 		EndpointName:   "pods",
@@ -129,96 +128,48 @@ func (everythingMatcher) MatchesSingle() (string, bool) {
 
 func TestEtcdList(t *testing.T) {
 	podA := &api.Pod{
-		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "foo"},
-		Spec:       api.PodSpec{NodeName: "machine"},
-	}
-	podB := &api.Pod{
 		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "bar"},
 		Spec:       api.PodSpec{NodeName: "machine"},
 	}
-
-	singleElemListResp := &etcd.Response{
-		Node: &etcd.Node{
-			Value: runtime.EncodeOrDie(testapi.Default.Codec(), podA),
-		},
-	}
-	normalListResp := &etcd.Response{
-		Node: &etcd.Node{
-			Nodes: []*etcd.Node{
-				{Value: runtime.EncodeOrDie(testapi.Default.Codec(), podA)},
-				{Value: runtime.EncodeOrDie(testapi.Default.Codec(), podB)},
-			},
-		},
+	podB := &api.Pod{
+		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "foo"},
+		Spec:       api.PodSpec{NodeName: "machine"},
 	}
 
 	testContext := api.WithNamespace(api.NewContext(), "test")
 	noNamespaceContext := api.NewContext()
 
 	table := map[string]struct {
-		in      tools.EtcdResponseWithError
+		in      *api.PodList
 		m       generic.Matcher
 		out     runtime.Object
 		context api.Context
-		succeed bool
 	}{
-		"empty": {
-			in: tools.EtcdResponseWithError{
-				R: &etcd.Response{
-					Node: &etcd.Node{
-						Nodes: []*etcd.Node{},
-					},
-				},
-				E: nil,
-			},
-			m:       everythingMatcher{},
-			out:     &api.PodList{Items: []api.Pod{}},
-			succeed: true,
-		},
 		"notFound": {
-			in: tools.EtcdResponseWithError{
-				R: &etcd.Response{},
-				E: tools.EtcdErrorNotFound,
-			},
-			m:       everythingMatcher{},
-			out:     &api.PodList{Items: []api.Pod{}},
-			succeed: true,
+			in:  nil,
+			m:   everythingMatcher{},
+			out: &api.PodList{Items: []api.Pod{}},
 		},
 		"normal": {
-			in: tools.EtcdResponseWithError{
-				R: normalListResp,
-				E: nil,
-			},
-			m:       everythingMatcher{},
-			out:     &api.PodList{Items: []api.Pod{*podA, *podB}},
-			succeed: true,
+			in:  &api.PodList{Items: []api.Pod{*podA, *podB}},
+			m:   everythingMatcher{},
+			out: &api.PodList{Items: []api.Pod{*podA, *podB}},
 		},
 		"normalFiltered": {
-			in: tools.EtcdResponseWithError{
-				R: singleElemListResp,
-				E: nil,
-			},
-			m:       setMatcher{sets.NewString("foo")},
-			out:     &api.PodList{Items: []api.Pod{*podA}},
-			succeed: true,
+			in:  &api.PodList{Items: []api.Pod{*podA, *podB}},
+			m:   setMatcher{sets.NewString("foo")},
+			out: &api.PodList{Items: []api.Pod{*podB}},
 		},
 		"normalFilteredNoNamespace": {
-			in: tools.EtcdResponseWithError{
-				R: normalListResp,
-				E: nil,
-			},
+			in:      &api.PodList{Items: []api.Pod{*podA, *podB}},
 			m:       setMatcher{sets.NewString("foo")},
-			out:     &api.PodList{Items: []api.Pod{*podA}},
+			out:     &api.PodList{Items: []api.Pod{*podB}},
 			context: noNamespaceContext,
-			succeed: true,
 		},
 		"normalFilteredMatchMultiple": {
-			in: tools.EtcdResponseWithError{
-				R: normalListResp,
-				E: nil,
-			},
-			m:       setMatcher{sets.NewString("foo", "makeMatchSingleReturnFalse")},
-			out:     &api.PodList{Items: []api.Pod{*podA}},
-			succeed: true,
+			in:  &api.PodList{Items: []api.Pod{*podA, *podB}},
+			m:   setMatcher{sets.NewString("foo", "makeMatchSingleReturnFalse")},
+			out: &api.PodList{Items: []api.Pod{*podB}},
 		},
 	}
 
@@ -227,29 +178,25 @@ func TestEtcdList(t *testing.T) {
 		if item.context != nil {
 			ctx = item.context
 		}
-		fakeClient, registry := NewTestGenericEtcdRegistry(t)
-		if name, ok := item.m.MatchesSingle(); ok {
-			if key, err := registry.KeyFunc(ctx, name); err == nil {
-				key = etcdtest.AddPrefix(key)
-				fakeClient.Data[key] = item.in
-			} else {
-				key := registry.KeyRootFunc(ctx)
-				key = etcdtest.AddPrefix(key)
-				fakeClient.Data[key] = item.in
+		server, registry := NewTestGenericEtcdRegistry(t)
+
+		if item.in != nil {
+			if err := storagetesting.CreateList(t, "/pods", registry.Storage, item.in); err != nil {
+				t.Errorf("Unexpected error %v", err)
 			}
-		} else {
-			key := registry.KeyRootFunc(ctx)
-			key = etcdtest.AddPrefix(key)
-			fakeClient.Data[key] = item.in
 		}
+
 		list, err := registry.ListPredicate(ctx, item.m, nil)
-		if e, a := item.succeed, err == nil; e != a {
-			t.Errorf("%v: expected %v, got %v: %v", name, e, a, err)
+		if err != nil {
+			t.Errorf("Unexpected error %v", err)
 			continue
 		}
+
+		// DeepDerivative e,a is needed here b/c the storage layer sets ResourceVersion
 		if e, a := item.out, list; !api.Semantic.DeepDerivative(e, a) {
 			t.Errorf("%v: Expected %#v, got %#v", name, e, a)
 		}
+		server.Terminate(t)
 	}
 }
 
@@ -263,78 +210,50 @@ func TestEtcdCreate(t *testing.T) {
 		Spec:       api.PodSpec{NodeName: "machine2"},
 	}
 
-	nodeWithPodA := tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podA),
-				ModifiedIndex: 1,
-				CreatedIndex:  1,
-			},
-		},
-		E: nil,
-	}
-
-	emptyNode := tools.EtcdResponseWithError{
-		R: &etcd.Response{},
-		E: tools.EtcdErrorNotFound,
-	}
-
 	testContext := api.WithNamespace(api.NewContext(), "test")
+	server, registry := NewTestGenericEtcdRegistry(t)
+	defer server.Terminate(t)
 
-	table := map[string]struct {
-		existing tools.EtcdResponseWithError
-		expect   tools.EtcdResponseWithError
-		toCreate runtime.Object
-		objOK    func(obj runtime.Object) bool
-		errOK    func(error) bool
-	}{
-		"normal": {
-			existing: emptyNode,
-			toCreate: podA,
-			objOK:    hasCreated(t, podA),
-			errOK:    func(err error) bool { return err == nil },
-		},
-		"preExisting": {
-			existing: nodeWithPodA,
-			expect:   nodeWithPodA,
-			toCreate: podB,
-			errOK:    errors.IsAlreadyExists,
-		},
+	// create the object
+	objA, err := registry.Create(testContext, podA)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
 	}
 
-	for name, item := range table {
-		fakeClient, registry := NewTestGenericEtcdRegistry(t)
-		path := etcdtest.AddPrefix("pods/foo")
-		fakeClient.Data[path] = item.existing
-		obj, err := registry.Create(testContext, item.toCreate)
-		if !item.errOK(err) {
-			t.Errorf("%v: unexpected error: %v", name, err)
-		}
+	// get the object
+	checkobj, err := registry.Get(testContext, podA.Name)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
 
-		actual := fakeClient.Data[path]
-		if item.objOK != nil {
-			if !item.objOK(obj) {
-				t.Errorf("%v: unexpected returned: %v", name, obj)
-			}
-			actualObj, err := api.Scheme.Decode([]byte(actual.R.Node.Value))
-			if err != nil {
-				t.Errorf("unable to decode stored value for %#v", actual)
-				continue
-			}
-			if !item.objOK(actualObj) {
-				t.Errorf("%v: unexpected response: %v", name, actual)
-			}
-		} else {
-			if e, a := item.expect, actual; !api.Semantic.DeepDerivative(e, a) {
-				t.Errorf("%v:\n%s", name, util.ObjectDiff(e, a))
-			}
-		}
+	// verify objects are equal
+	if e, a := objA, checkobj; !reflect.DeepEqual(e, a) {
+		t.Errorf("Expected %#v, got %#v", e, a)
+	}
+
+	// now try to create the second pod
+	_, err = registry.Create(testContext, podB)
+	if !errors.IsAlreadyExists(err) {
+		t.Errorf("Unexpected error: %v", err)
 	}
 }
 
-func podCopy(in *api.Pod) *api.Pod {
-	out := *in
-	return &out
+func updateAndVerify(t *testing.T, ctx api.Context, registry *Etcd, pod *api.Pod) bool {
+	obj, _, err := registry.Update(ctx, pod)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+		return false
+	}
+	checkObj, err := registry.Get(ctx, pod.Name)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+		return false
+	}
+	if e, a := obj, checkObj; !reflect.DeepEqual(e, a) {
+		t.Errorf("Expected %#v, got %#v", e, a)
+		return false
+	}
+	return true
 }
 
 func TestEtcdUpdate(t *testing.T) {
@@ -343,254 +262,106 @@ func TestEtcdUpdate(t *testing.T) {
 		Spec:       api.PodSpec{NodeName: "machine"},
 	}
 	podB := &api.Pod{
-		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "test", ResourceVersion: "1"},
+		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "test"},
 		Spec:       api.PodSpec{NodeName: "machine2"},
 	}
 	podAWithResourceVersion := &api.Pod{
-		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "test", ResourceVersion: "3"},
+		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "test", ResourceVersion: "7"},
 		Spec:       api.PodSpec{NodeName: "machine"},
 	}
-	nodeWithPodA := tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podA),
-				ModifiedIndex: 1,
-				CreatedIndex:  1,
-			},
-		},
-		E: nil,
-	}
-
-	newerNodeWithPodA := tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podA),
-				ModifiedIndex: 2,
-				CreatedIndex:  1,
-			},
-		},
-		E: nil,
-	}
-
-	nodeWithPodB := tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podB),
-				ModifiedIndex: 1,
-				CreatedIndex:  1,
-			},
-		},
-		E: nil,
-	}
-
-	nodeWithPodAWithResourceVersion := tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podAWithResourceVersion),
-				ModifiedIndex: 3,
-				CreatedIndex:  1,
-			},
-		},
-		E: nil,
-	}
-	emptyNode := tools.EtcdResponseWithError{
-		R: &etcd.Response{},
-		E: tools.EtcdErrorNotFound,
-	}
 
 	testContext := api.WithNamespace(api.NewContext(), "test")
+	server, registry := NewTestGenericEtcdRegistry(t)
+	defer server.Terminate(t)
 
-	table := map[string]struct {
-		existing                 tools.EtcdResponseWithError
-		expect                   tools.EtcdResponseWithError
-		toUpdate                 runtime.Object
-		allowCreate              bool
-		allowUnconditionalUpdate bool
-		objOK                    func(obj runtime.Object) bool
-		errOK                    func(error) bool
-	}{
-		"normal": {
-			existing: nodeWithPodA,
-			expect:   nodeWithPodB,
-			toUpdate: podCopy(podB),
-			errOK:    func(err error) bool { return err == nil },
-		},
-		"notExisting": {
-			existing: emptyNode,
-			expect:   emptyNode,
-			toUpdate: podCopy(podA),
-			errOK:    func(err error) bool { return errors.IsNotFound(err) },
-		},
-		"createIfNotFound": {
-			existing:    emptyNode,
-			toUpdate:    podCopy(podA),
-			allowCreate: true,
-			objOK:       hasCreated(t, podA),
-			errOK:       func(err error) bool { return err == nil },
-		},
-		"outOfDate": {
-			existing: newerNodeWithPodA,
-			expect:   newerNodeWithPodA,
-			toUpdate: podCopy(podB),
-			errOK:    func(err error) bool { return errors.IsConflict(err) },
-		},
-		"unconditionalUpdate": {
-			existing:                 nodeWithPodAWithResourceVersion,
-			allowUnconditionalUpdate: true,
-			toUpdate:                 podCopy(podA),
-			objOK:                    func(obj runtime.Object) bool { return true },
-			errOK:                    func(err error) bool { return err == nil },
-		},
+	// Test1 try to update a non-existing node
+	_, _, err := registry.Update(testContext, podA)
+	if !errors.IsNotFound(err) {
+		t.Errorf("Unexpected error: %v", err)
 	}
 
-	for name, item := range table {
-		fakeClient, registry := NewTestGenericEtcdRegistry(t)
-		registry.UpdateStrategy.(*testRESTStrategy).allowCreateOnUpdate = item.allowCreate
-		registry.UpdateStrategy.(*testRESTStrategy).allowUnconditionalUpdate = item.allowUnconditionalUpdate
-		path := etcdtest.AddPrefix("pods/foo")
-		fakeClient.Data[path] = item.existing
-		obj, _, err := registry.Update(testContext, item.toUpdate)
-		if !item.errOK(err) {
-			t.Errorf("%v: unexpected error: %v", name, err)
-		}
-
-		actual := fakeClient.Data[path]
-		if item.objOK != nil {
-			if !item.objOK(obj) {
-				t.Errorf("%v: unexpected returned: %#v", name, obj)
-			}
-			actualObj, err := api.Scheme.Decode([]byte(actual.R.Node.Value))
-			if err != nil {
-				t.Errorf("unable to decode stored value for %#v", actual)
-				continue
-			}
-			if !item.objOK(actualObj) {
-				t.Errorf("%v: unexpected response: %#v", name, actual)
-			}
-		} else {
-			if e, a := item.expect, actual; !api.Semantic.DeepDerivative(e, a) {
-				t.Errorf("%v:\n%s", name, util.ObjectDiff(e, a))
-			}
-		}
+	// Test2 createIfNotFound and verify
+	registry.UpdateStrategy.(*testRESTStrategy).allowCreateOnUpdate = true
+	if !updateAndVerify(t, testContext, registry, podA) {
+		t.Errorf("Unexpected error updating podA")
 	}
+	registry.UpdateStrategy.(*testRESTStrategy).allowCreateOnUpdate = false
+
+	// Test3 outofDate
+	_, _, err = registry.Update(testContext, podAWithResourceVersion)
+	if !errors.IsConflict(err) {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// Test4 normal update and verify
+	if !updateAndVerify(t, testContext, registry, podB) {
+		t.Errorf("Unexpected error updating podB")
+	}
+
+	// Test5 unconditional update
+	// NOTE: The logic for unconditional updates doesn't make sense to me, and imho should be removed.
+	// doUnconditionalUpdate := resourceVersion == 0 && e.UpdateStrategy.AllowUnconditionalUpdate()
+	// ^^ That condition can *never be true due to the creation of root objects.
+	//
+	// registry.UpdateStrategy.(*testRESTStrategy).allowUnconditionalUpdate = true
+	// updateAndVerify(t, testContext, registry, podAWithResourceVersion)
+
 }
 
 func TestEtcdGet(t *testing.T) {
 	podA := &api.Pod{
-		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "foo", ResourceVersion: "1"},
+		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "foo"},
 		Spec:       api.PodSpec{NodeName: "machine"},
 	}
 
-	nodeWithPodA := tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podA),
-				ModifiedIndex: 1,
-				CreatedIndex:  1,
-			},
-		},
-		E: nil,
-	}
-
-	emptyNode := tools.EtcdResponseWithError{
-		R: &etcd.Response{},
-		E: tools.EtcdErrorNotFound,
-	}
-
-	key := "foo"
-
-	table := map[string]struct {
-		existing tools.EtcdResponseWithError
-		expect   runtime.Object
-		errOK    func(error) bool
-	}{
-		"normal": {
-			existing: nodeWithPodA,
-			expect:   podA,
-			errOK:    func(err error) bool { return err == nil },
-		},
-		"notExisting": {
-			existing: emptyNode,
-			expect:   nil,
-			errOK:    errors.IsNotFound,
-		},
-	}
-
 	testContext := api.WithNamespace(api.NewContext(), "test")
+	server, registry := NewTestGenericEtcdRegistry(t)
+	defer server.Terminate(t)
 
-	for name, item := range table {
-		fakeClient, registry := NewTestGenericEtcdRegistry(t)
-		path := etcdtest.AddPrefix("pods/foo")
-		fakeClient.Data[path] = item.existing
-		got, err := registry.Get(testContext, key)
-		if !item.errOK(err) {
-			t.Errorf("%v: unexpected error: %v", name, err)
-		}
-
-		if e, a := item.expect, got; !api.Semantic.DeepDerivative(e, a) {
-			t.Errorf("%v:\n%s", name, util.ObjectDiff(e, a))
-		}
+	_, err := registry.Get(testContext, podA.Name)
+	if !errors.IsNotFound(err) {
+		t.Errorf("Unexpected error: %v", err)
 	}
+
+	registry.UpdateStrategy.(*testRESTStrategy).allowCreateOnUpdate = true
+	if !updateAndVerify(t, testContext, registry, podA) {
+		t.Errorf("Unexpected error updating podA")
+	}
+
 }
 
 func TestEtcdDelete(t *testing.T) {
 	podA := &api.Pod{
-		ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: "1"},
+		ObjectMeta: api.ObjectMeta{Name: "foo"},
 		Spec:       api.PodSpec{NodeName: "machine"},
 	}
 
-	nodeWithPodA := tools.EtcdResponseWithError{
-		R: &etcd.Response{
-			Node: &etcd.Node{
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podA),
-				ModifiedIndex: 1,
-				CreatedIndex:  1,
-			},
-		},
-		E: nil,
-	}
-
-	emptyNode := tools.EtcdResponseWithError{
-		R: &etcd.Response{},
-		E: tools.EtcdErrorNotFound,
-	}
-
 	testContext := api.WithNamespace(api.NewContext(), "test")
+	server, registry := NewTestGenericEtcdRegistry(t)
+	defer server.Terminate(t)
 
-	key := "foo"
-
-	table := map[string]struct {
-		existing tools.EtcdResponseWithError
-		expect   tools.EtcdResponseWithError
-		errOK    func(error) bool
-	}{
-		"normal": {
-			existing: nodeWithPodA,
-			expect:   emptyNode,
-			errOK:    func(err error) bool { return err == nil },
-		},
-		"notExisting": {
-			existing: emptyNode,
-			expect:   emptyNode,
-			errOK:    func(err error) bool { return errors.IsNotFound(err) },
-		},
+	// test failure condition
+	_, err := registry.Delete(testContext, podA.Name, nil)
+	if !errors.IsNotFound(err) {
+		t.Errorf("Unexpected error: %v", err)
 	}
 
-	for name, item := range table {
-		fakeClient, registry := NewTestGenericEtcdRegistry(t)
-		path := etcdtest.AddPrefix("pods/foo")
-		fakeClient.Data[path] = item.existing
-		obj, err := registry.Delete(testContext, key, nil)
-		if !item.errOK(err) {
-			t.Errorf("%v: unexpected error: %v (%#v)", name, err, obj)
-		}
+	// create pod
+	_, err = registry.Create(testContext, podA)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
 
-		if item.expect.E != nil {
-			item.expect.E.(*etcd.EtcdError).Index = fakeClient.ChangeIndex
-		}
-		if e, a := item.expect, fakeClient.Data[path]; !api.Semantic.DeepDerivative(e, a) {
-			t.Errorf("%v:\n%s", name, util.ObjectDiff(e, a))
-		}
+	// delete object
+	_, err = registry.Delete(testContext, podA.Name, nil)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// try to get a item which should be deleted
+	_, err = registry.Get(testContext, podA.Name)
+	if !errors.IsNotFound(err) {
+		t.Errorf("Unexpected error: %v", err)
 	}
 }
 
@@ -621,42 +392,30 @@ func TestEtcdWatch(t *testing.T) {
 		}
 		podA := &api.Pod{
 			ObjectMeta: api.ObjectMeta{
-				Name:            "foo",
-				Namespace:       "test",
-				ResourceVersion: "1",
+				Name:      "foo",
+				Namespace: "test",
 			},
 			Spec: api.PodSpec{NodeName: "machine"},
 		}
-		respWithPodA := &etcd.Response{
-			Node: &etcd.Node{
-				Key:           "/registry/pods/test/foo",
-				Value:         runtime.EncodeOrDie(testapi.Default.Codec(), podA),
-				ModifiedIndex: 1,
-				CreatedIndex:  1,
-			},
-			Action: "create",
-		}
 
-		fakeClient, registry := NewTestGenericEtcdRegistry(t)
-		wi, err := registry.WatchPredicate(ctx, m, "1")
+		server, registry := NewTestGenericEtcdRegistry(t)
+		wi, err := registry.WatchPredicate(ctx, m, "0")
 		if err != nil {
 			t.Errorf("%v: unexpected error: %v", name, err)
-			continue
-		}
-		fakeClient.WaitForWatchCompletion()
-
-		go func() {
-			fakeClient.WatchResponse <- respWithPodA
-		}()
-
-		got, open := <-wi.ResultChan()
-		if !open {
-			t.Errorf("%v: unexpected channel close", name)
-			continue
+		} else {
+			obj, err := registry.Create(testContext, podA)
+			if err != nil {
+				got, open := <-wi.ResultChan()
+				if !open {
+					t.Errorf("%v: unexpected channel close", name)
+				} else {
+					if e, a := obj, got.Object; !reflect.DeepEqual(e, a) {
+						t.Errorf("Expected %#v, got %#v", e, a)
+					}
+				}
+			}
 		}
 
-		if e, a := podA, got.Object; !api.Semantic.DeepDerivative(e, a) {
-			t.Errorf("%v: difference: %s", name, util.ObjectDiff(e, a))
-		}
+		server.Terminate(t)
 	}
 }

pkg/storage/testing/utils.go

@@ -0,0 +1,57 @@
+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package testing
+
+import (
+	"path"
+	"testing"
+
+	"golang.org/x/net/context"
+	"k8s.io/kubernetes/pkg/api/meta"
+	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/storage"
+)
+
+// CreateObj will create a single object using the storage interface
+func CreateObj(t *testing.T, helper storage.Interface, name string, obj, out runtime.Object, ttl uint64) error {
+	err := helper.Set(context.TODO(), name, obj, out, ttl)
+	if err != nil {
+		t.Errorf("Unexpected error %v", err)
+	}
+	return err
+}
+
+// CreateList will properly create a list using the storage interface
+func CreateList(t *testing.T, prefix string, helper storage.Interface, list runtime.Object) error {
+	items, err := runtime.ExtractList(list)
+	if err != nil {
+		return err
+	}
+	for i := range items {
+		obj := items[i]
+		meta, err := meta.Accessor(obj)
+		if err != nil {
+			return err
+		}
+		err = CreateObj(t, helper, path.Join(prefix, meta.Name()), obj, obj, 0)
+		if err != nil {
+			return err
+		}
+		items[i] = obj
+	}
+	return runtime.SetList(list, items)
+}