e2e: add test for v3 watch over grpc gateway

e2e/v2_curl_test.go

@@ -120,7 +120,8 @@ type cURLReq struct {
 	username string
 	password string
 
-	isTLS bool
+	isTLS   bool
+	timeout int
 
 	endpoint string
 
@@ -151,6 +152,9 @@ func cURLPrefixArgs(clus *etcdProcessCluster, method string, req cURLReq) []stri
 	} else {
 		cmdArgs = append(cmdArgs, "-L", ep)
 	}
+	if req.timeout != 0 {
+		cmdArgs = append(cmdArgs, "-m", fmt.Sprintf("%d", req.timeout))
+	}
 
 	switch method {
 	case "POST", "PUT":

e2e/v3_curl_test.go

@@ -18,7 +18,7 @@ import (
 	"encoding/json"
 	"testing"
 
-	"github.com/coreos/etcd/etcdserver/etcdserverpb"
+	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"github.com/coreos/etcd/pkg/testutil"
 )
 
@@ -47,14 +47,14 @@ func testCurlPutGetGRPCGateway(t *testing.T, cfg *etcdProcessClusterConfig) {
 		expectPut = `"revision":"`
 		expectGet = `"value":"`
 	)
-	putData, err := json.Marshal(&etcdserverpb.PutRequest{
+	putData, err := json.Marshal(&pb.PutRequest{
 		Key:   key,
 		Value: value,
 	})
 	if err != nil {
 		t.Fatal(err)
 	}
-	rangeData, err := json.Marshal(&etcdserverpb.RangeRequest{
+	rangeData, err := json.Marshal(&pb.RangeRequest{
 		Key: key,
 	})
 	if err != nil {
@@ -74,3 +74,40 @@ func testCurlPutGetGRPCGateway(t *testing.T, cfg *etcdProcessClusterConfig) {
 		}
 	}
 }
+
+func TestV3CurlWatch(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	epc, err := newEtcdProcessCluster(&configNoTLS)
+	if err != nil {
+		t.Fatalf("could not start etcd process cluster (%v)", err)
+	}
+	defer func() {
+		if cerr := epc.Close(); err != nil {
+			t.Fatalf("error closing etcd processes (%v)", cerr)
+		}
+	}()
+
+	// store "bar" into "foo"
+	putreq, err := json.Marshal(&pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = cURLPost(epc, cURLReq{endpoint: "/v3alpha/kv/put", value: string(putreq), expected: "revision"}); err != nil {
+		t.Fatalf("failed put with curl (%v)", err)
+	}
+	// watch for first update to "foo"
+	wcr := &pb.WatchCreateRequest{Key: []byte("foo"), StartRevision: 1}
+	wreq, err := json.Marshal(wcr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// marshaling the grpc to json gives:
+	// "{"RequestUnion":{"CreateRequest":{"key":"Zm9v","start_revision":1}}}"
+	// but the gprc-gateway expects a different format..
+	wstr := `{"create_request" : ` + string(wreq) + "}"
+	// expects "bar", timeout after 2 seconds since stream waits forever
+	if err = cURLPost(epc, cURLReq{endpoint: "/v3alpha/watch", value: wstr, expected: `"YmFy"`, timeout: 2}); err != nil {
+		t.Fatal(err)
+	}
+}