func TestV3WatchCancelSynced(t *testing.T) {
	defer testutil.AfterTest(t)
	testV3WatchCancel(t, 0)
}
TestV3WatchCancelSynced tests Watch APIs cancellation from synced map.
package integration
import (
	"bytes"
	"fmt"
	"reflect"
	"sort"
	"sync"
	"testing"
	"time"
	"github.com/coreos/etcd/etcdserver/api/v3rpc"
	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/coreos/etcd/pkg/testutil"
	"golang.org/x/net/context"
)
func TestV3WatchFromCurrentRevision(t *testing.T) {
	defer testutil.AfterTest(t)
	tests := []struct {
		putKeys      []string
		watchRequest *pb.WatchRequest
		wresps []*pb.WatchResponse
	}{
		{
			[]string{"foo"},
			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
				CreateRequest: &pb.WatchCreateRequest{
					Key: []byte("foo")}}},
			[]*pb.WatchResponse{
				{
					Header:  &pb.ResponseHeader{Revision: 2},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1},
						},
					},
				},
			},
		},
		{
			[]string{"foo"},
			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
				CreateRequest: &pb.WatchCreateRequest{
					Key: []byte("helloworld")}}},
			[]*pb.WatchResponse{},
		},
		{
			[]string{"fooLong"},
			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
				CreateRequest: &pb.WatchCreateRequest{
					Key:      []byte("foo"),
					RangeEnd: []byte("fop")}}},
			[]*pb.WatchResponse{
				{
					Header:  &pb.ResponseHeader{Revision: 2},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("fooLong"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1},
						},
					},
				},
			},
		},
		{
			[]string{"foo"},
			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
				CreateRequest: &pb.WatchCreateRequest{
					Key:      []byte("helloworld"),
					RangeEnd: []byte("helloworle")}}},
			[]*pb.WatchResponse{},
		},
		{
			[]string{"fooLong"},
			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
				CreateRequest: &pb.WatchCreateRequest{
					Key:      []byte(""),
					RangeEnd: []byte("\x00")}}},
			[]*pb.WatchResponse{
				{
					Header:  &pb.ResponseHeader{Revision: 2},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("fooLong"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1},
						},
					},
				},
			},
		},
		{
			[]string{"foo", "foo", "foo"},
			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
				CreateRequest: &pb.WatchCreateRequest{
					Key: []byte("foo")}}},
			[]*pb.WatchResponse{
				{
					Header:  &pb.ResponseHeader{Revision: 2},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1},
						},
					},
				},
				{
					Header:  &pb.ResponseHeader{Revision: 3},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 3, Version: 2},
						},
					},
				},
				{
					Header:  &pb.ResponseHeader{Revision: 4},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 4, Version: 3},
						},
					},
				},
			},
		},
		{
			[]string{"foo", "foo", "foo"},
			&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
				CreateRequest: &pb.WatchCreateRequest{
					Key:      []byte("foo"),
					RangeEnd: []byte("fop")}}},
			[]*pb.WatchResponse{
				{
					Header:  &pb.ResponseHeader{Revision: 2},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1},
						},
					},
				},
				{
					Header:  &pb.ResponseHeader{Revision: 3},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 3, Version: 2},
						},
					},
				},
				{
					Header:  &pb.ResponseHeader{Revision: 4},
					Created: false,
					Events: []*mvccpb.Event{
						{
							Type: mvccpb.PUT,
							Kv:   &mvccpb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 4, Version: 3},
						},
					},
				},
			},
		},
	}
	for i, tt := range tests {
		clus := NewClusterV3(t, &ClusterConfig{Size: 3})
		wAPI := toGRPC(clus.RandClient()).Watch
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		wStream, err := wAPI.Watch(ctx)
		if err != nil {
			t.Fatalf("#%d: wAPI.Watch error: %v", i, err)
		}
		err = wStream.Send(tt.watchRequest)
		if err != nil {
			t.Fatalf("#%d: wStream.Send error: %v", i, err)
		}
		cresp, err := wStream.Recv()
		if err != nil {
			t.Errorf("#%d: wStream.Recv error: %v", i, err)
			continue
		}
		if !cresp.Created {
			t.Errorf("#%d: did not create watchid, got %+v", i, cresp)
			continue
		}
		if cresp.Canceled {
			t.Errorf("#%d: canceled watcher on create %+v", i, cresp)
			continue
		}
		createdWatchId := cresp.WatchId
		if cresp.Header == nil || cresp.Header.Revision != 1 {
			t.Errorf("#%d: header revision got +%v, wanted revison 1", i, cresp)
			continue
		}
		go func() {
			for _, k := range tt.putKeys {
				kvc := toGRPC(clus.RandClient()).KV
				req := &pb.PutRequest{Key: []byte(k), Value: []byte("bar")}
				if _, err := kvc.Put(context.TODO(), req); err != nil {
					t.Fatalf("#%d: couldn't put key (%v)", i, err)
				}
			}
		}()
		for j, wresp := range tt.wresps {
			resp, err := wStream.Recv()
			if err != nil {
				t.Errorf("#%d.%d: wStream.Recv error: %v", i, j, err)
			}
			if resp.Header == nil {
				t.Fatalf("#%d.%d: unexpected nil resp.Header", i, j)
			}
			if resp.Header.Revision != wresp.Header.Revision {
				t.Errorf("#%d.%d: resp.Header.Revision got = %d, want = %d", i, j, resp.Header.Revision, wresp.Header.Revision)
			}
			if wresp.Created != resp.Created {
				t.Errorf("#%d.%d: resp.Created got = %v, want = %v", i, j, resp.Created, wresp.Created)
			}
			if resp.WatchId != createdWatchId {
				t.Errorf("#%d.%d: resp.WatchId got = %d, want = %d", i, j, resp.WatchId, createdWatchId)
			}
			if !reflect.DeepEqual(resp.Events, wresp.Events) {
				t.Errorf("#%d.%d: resp.Events got = %+v, want = %+v", i, j, resp.Events, wresp.Events)
			}
		}
		rok, nr := waitResponse(wStream, 1*time.Second)
		if !rok {
			t.Errorf("unexpected pb.WatchResponse is received %+v", nr)
		}
		clus.Terminate(t)
	}
}
func TestV3WatchFutureRevision(t *testing.T) {
	defer testutil.AfterTest(t)
	clus := NewClusterV3(t, &ClusterConfig{Size: 1})
	defer clus.Terminate(t)
	wAPI := toGRPC(clus.RandClient()).Watch
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	wStream, err := wAPI.Watch(ctx)
	if err != nil {
		t.Fatalf("wAPI.Watch error: %v", err)
	}
	wkey := []byte("foo")
	wrev := int64(10)
	req := &pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
		CreateRequest: &pb.WatchCreateRequest{Key: wkey, StartRevision: wrev}}}
	err = wStream.Send(req)
	if err != nil {
		t.Fatalf("wStream.Send error: %v", err)
	}
	cresp, err := wStream.Recv()
	if err != nil {
		t.Fatalf("wStream.Recv error: %v", err)
	}
	if !cresp.Created {
		t.Fatalf("create %v, want %v", cresp.Created, true)
	}
	kvc := toGRPC(clus.RandClient()).KV
	for {
		req := &pb.PutRequest{Key: wkey, Value: []byte("bar")}
		resp, rerr := kvc.Put(context.TODO(), req)
		if rerr != nil {
			t.Fatalf("couldn't put key (%v)", rerr)
		}
		if resp.Header.Revision == wrev {
			break
		}
	}
	cresp, err = wStream.Recv()
	if err != nil {
		t.Fatalf("wStream.Recv error: %v", err)
	}
	if cresp.Header.Revision != wrev {
		t.Fatalf("revision = %d, want %d", cresp.Header.Revision, wrev)
	}
	if len(cresp.Events) != 1 {
		t.Fatalf("failed to receive events")
	}
	if cresp.Events[0].Kv.ModRevision != wrev {
		t.Errorf("mod revision = %d, want %d", cresp.Events[0].Kv.ModRevision, wrev)
	}
}
func TestV3WatchWrongRange(t *testing.T) {
	defer testutil.AfterTest(t)
	clus := NewClusterV3(t, &ClusterConfig{Size: 1})
	defer clus.Terminate(t)
	wAPI := toGRPC(clus.RandClient()).Watch
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	wStream, err := wAPI.Watch(ctx)
	if err != nil {
		t.Fatalf("wAPI.Watch error: %v", err)
	}
	tests := []struct {
		key      []byte
		end      []byte
		canceled bool
	}{
		{[]byte("a"), []byte("a"), true},   
		{[]byte("b"), []byte("a"), true},   
		{[]byte("foo"), []byte{0}, false},  
	}
	for i, tt := range tests {
		if err := wStream.Send(&pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{
			CreateRequest: &pb.WatchCreateRequest{Key: tt.key, RangeEnd: tt.end, StartRevision: 1}}}); err != nil {
			t.Fatalf("#%d: wStream.Send error: %v", i, err)
		}
		cresp, err := wStream.Recv()
		if err != nil {
			t.Fatalf("#%d: wStream.Recv error: %v", i, err)
		}
		if !cresp.Created {
			t.Fatalf("#%d: create %v, want %v", i, cresp.Created, true)
		}
		if cresp.Canceled != tt.canceled {
			t.Fatalf("#%d: canceled %v, want %v", i, tt.canceled, cresp.Canceled)
		}
		if tt.canceled && cresp.WatchId != -1 {
			t.Fatalf("#%d: canceled watch ID %d, want -1", i, cresp.WatchId)
		}
	}
}