Merge pull request #1762 from lavalamp/fix

Fix missing error type in watch decode.

pkg/api/testapi/testapi.go

@@ -24,7 +24,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 )
 
-// Version returns the API version to test against as set by the KUBE_API_VERSION env var.
+// Version returns the API version to test against, as set by the KUBE_API_VERSION env var.
 func Version() string {
 	version := os.Getenv("KUBE_API_VERSION")
 	if version == "" {
@@ -33,10 +33,22 @@ func Version() string {
 	return version
 }
 
-func CodecForVersionOrDie() runtime.Codec {
+// Codec returns the codec for the API version to test against, as set by the
+// KUBE_API_VERSION env var.
+func Codec() runtime.Codec {
 	interfaces, err := latest.InterfacesFor(Version())
 	if err != nil {
 		panic(err)
 	}
 	return interfaces.Codec
 }
+
+// ResourceVersioner returns the ResourceVersioner for the API version to test against,
+// as set by the KUBE_API_VERSION env var.
+func ResourceVersioner() runtime.ResourceVersioner {
+	interfaces, err := latest.InterfacesFor(Version())
+	if err != nil {
+		panic(err)
+	}
+	return interfaces.ResourceVersioner
+}

pkg/client/restclient_test.go

@@ -109,7 +109,7 @@ func TestDoRequest(t *testing.T) {
 		{Request: testRequest{Method: "GET", Path: "error"}, Response: Response{StatusCode: 500}, Error: true},
 		{Request: testRequest{Method: "POST", Path: "faildecode"}, Response: Response{StatusCode: 200, RawBody: &invalid}},
 		{Request: testRequest{Method: "GET", Path: "failread"}, Response: Response{StatusCode: 200, RawBody: &invalid}},
-		{Client: &Client{&RESTClient{baseURL: uri, Codec: testapi.CodecForVersionOrDie()}}, Request: testRequest{Method: "GET", Path: "nocertificate"}, Error: true},
+		{Client: &Client{&RESTClient{baseURL: uri, Codec: testapi.Codec()}}, Request: testRequest{Method: "GET", Path: "nocertificate"}, Error: true},
 	}
 	for _, c := range testClients {
 		client := c.Setup()

pkg/controller/replication_controller_test.go

@@ -149,7 +149,7 @@ func TestSyncReplicationControllerDeletes(t *testing.T) {
 }
 
 func TestSyncReplicationControllerCreates(t *testing.T) {
-	body := runtime.EncodeOrDie(testapi.CodecForVersionOrDie(), newPodList(0))
+	body := runtime.EncodeOrDie(testapi.Codec(), newPodList(0))
 	fakeHandler := util.FakeHandler{
 		StatusCode:   200,
 		ResponseBody: string(body),
@@ -170,7 +170,7 @@ func TestSyncReplicationControllerCreates(t *testing.T) {
 
 func TestCreateReplica(t *testing.T) {
 	ctx := api.NewDefaultContext()
-	body := runtime.EncodeOrDie(testapi.CodecForVersionOrDie(), &api.Pod{})
+	body := runtime.EncodeOrDie(testapi.Codec(), &api.Pod{})
 	fakeHandler := util.FakeHandler{
 		StatusCode:   200,
 		ResponseBody: string(body),

pkg/service/endpoints_controller_test.go

@@ -203,7 +203,7 @@ func TestSyncEndpointsItemsPreexisting(t *testing.T) {
 	if err := endpoints.SyncServiceEndpoints(); err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
-	data := runtime.EncodeOrDie(testapi.CodecForVersionOrDie(), &api.Endpoints{
+	data := runtime.EncodeOrDie(testapi.Codec(), &api.Endpoints{
 		TypeMeta: api.TypeMeta{
 			ID:              "foo",
 			ResourceVersion: "1",
@@ -261,7 +261,7 @@ func TestSyncEndpointsItems(t *testing.T) {
 	if err := endpoints.SyncServiceEndpoints(); err != nil {
 		t.Errorf("unexpected error: %v", err)
 	}
-	data := runtime.EncodeOrDie(testapi.CodecForVersionOrDie(), &api.Endpoints{
+	data := runtime.EncodeOrDie(testapi.Codec(), &api.Endpoints{
 		TypeMeta: api.TypeMeta{
 			ResourceVersion: "",
 		},

pkg/watch/json/decoder.go

@@ -51,7 +51,7 @@ func (d *Decoder) Decode() (watch.EventType, runtime.Object, error) {
 		return "", nil, err
 	}
 	switch got.Type {
-	case watch.Added, watch.Modified, watch.Deleted:
+	case watch.Added, watch.Modified, watch.Deleted, watch.Error:
 	default:
 		return "", nil, fmt.Errorf("got invalid watch event type: %v", got.Type)
 	}

pkg/watch/json/decoder_test.go

@@ -24,61 +24,65 @@ import (
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/v1beta1"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/testapi"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 )
 
 func TestDecoder(t *testing.T) {
-	out, in := io.Pipe()
+	table := []watch.EventType{watch.Added, watch.Deleted, watch.Modified, watch.Error}
-	decoder := NewDecoder(out, v1beta1.Codec)
 
-	expect := &api.Pod{TypeMeta: api.TypeMeta{ID: "foo"}}
+	for _, eventType := range table {
-	encoder := json.NewEncoder(in)
+		out, in := io.Pipe()
-	go func() {
+		decoder := NewDecoder(out, testapi.Codec())
-		data, err := v1beta1.Codec.Encode(expect)
-		if err != nil {
-			t.Fatalf("Unexpected error %v", err)
-		}
-		if err := encoder.Encode(&watchEvent{watch.Added, runtime.RawExtension{json.RawMessage(data)}}); err != nil {
-			t.Errorf("Unexpected error %v", err)
-		}
-		in.Close()
-	}()
 
-	done := make(chan struct{})
+		expect := &api.Pod{TypeMeta: api.TypeMeta{ID: "foo"}}
-	go func() {
+		encoder := json.NewEncoder(in)
-		action, got, err := decoder.Decode()
+		go func() {
-		if err != nil {
+			data, err := testapi.Codec().Encode(expect)
-			t.Fatalf("Unexpected error %v", err)
+			if err != nil {
-		}
+				t.Fatalf("Unexpected error %v", err)
-		if e, a := watch.Added, action; e != a {
+			}
-			t.Errorf("Expected %v, got %v", e, a)
+			if err := encoder.Encode(&watchEvent{eventType, runtime.RawExtension{json.RawMessage(data)}}); err != nil {
-		}
+				t.Errorf("Unexpected error %v", err)
-		if e, a := expect, got; !reflect.DeepEqual(e, a) {
+			}
-			t.Errorf("Expected %v, got %v", e, a)
+			in.Close()
-		}
+		}()
-		t.Logf("Exited read")
-		close(done)
-	}()
-	<-done
 
-	done = make(chan struct{})
+		done := make(chan struct{})
-	go func() {
+		go func() {
-		_, _, err := decoder.Decode()
+			action, got, err := decoder.Decode()
-		if err == nil {
+			if err != nil {
-			t.Errorf("Unexpected nil error")
+				t.Fatalf("Unexpected error %v", err)
-		}
+			}
-		close(done)
+			if e, a := eventType, action; e != a {
-	}()
+				t.Errorf("Expected %v, got %v", e, a)
-	<-done
+			}
+			if e, a := expect, got; !reflect.DeepEqual(e, a) {
+				t.Errorf("Expected %v, got %v", e, a)
+			}
+			t.Logf("Exited read")
+			close(done)
+		}()
+		<-done
 
-	decoder.Close()
+		done = make(chan struct{})
+		go func() {
+			_, _, err := decoder.Decode()
+			if err == nil {
+				t.Errorf("Unexpected nil error")
+			}
+			close(done)
+		}()
+		<-done
+
+		decoder.Close()
+	}
 }
 
 func TestDecoder_SourceClose(t *testing.T) {
 	out, in := io.Pipe()
-	decoder := NewDecoder(out, v1beta1.Codec)
+	decoder := NewDecoder(out, testapi.Codec())
 
 	done := make(chan struct{})
 

plugin/pkg/scheduler/factory/factory_test.go

@@ -317,7 +317,7 @@ func TestBind(t *testing.T) {
 			t.Errorf("Unexpected error: %v", err)
 			continue
 		}
-		expectedBody := runtime.EncodeOrDie(testapi.CodecForVersionOrDie(), item.binding)
+		expectedBody := runtime.EncodeOrDie(testapi.Codec(), item.binding)
 		handler.ValidateRequest(t, "/api/"+testapi.Version()+"/bindings", "POST", &expectedBody)
 	}
 }