From 96f646c58681cd64530a36ed7d99a1ab040f9cd8 Mon Sep 17 00:00:00 2001 From: Gyu-Ho Lee Date: Wed, 20 Jan 2016 12:13:58 -0800 Subject: [PATCH] integration: add more tests for unsynced watch For https://github.com/coreos/etcd/issues/4216. --- integration/v3_grpc_test.go | 45 ++++++++++++++++++++++++++++--------- 1 file changed, 34 insertions(+), 11 deletions(-) diff --git a/integration/v3_grpc_test.go b/integration/v3_grpc_test.go index 0e08a0e5c..6261873a4 100644 --- a/integration/v3_grpc_test.go +++ b/integration/v3_grpc_test.go @@ -483,11 +483,19 @@ func testV3WatchCancel(t *testing.T, startRev int64) { clus.Terminate(t) } -// TestV3WatchMultiple tests multiple watchers on the same key +func TestV3WatchMultipleWatchersSynced(t *testing.T) { + testV3WatchMultipleWatchers(t, 0) +} + +func TestV3WatchMultipleWatchersUnsynced(t *testing.T) { + testV3WatchMultipleWatchers(t, 1) +} + +// testV3WatchMultipleWatchers tests multiple watchers on the same key // and one watcher with matching prefix. It first puts the key // that matches all watchers, and another key that matches only // one watcher to test if it receives expected events. -func TestV3WatchMultiple(t *testing.T) { +func testV3WatchMultipleWatchers(t *testing.T, startRev int64) { clus := newClusterGRPC(t, &clusterConfig{size: 3}) wAPI := pb.NewWatchClient(clus.RandConn()) kvc := pb.NewKVClient(clus.RandConn()) @@ -501,9 +509,9 @@ func TestV3WatchMultiple(t *testing.T) { for i := 0; i < watchKeyN+1; i++ { var wreq *pb.WatchRequest if i < watchKeyN { - wreq = &pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo")}} + wreq = &pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo"), StartRevision: startRev}} } else { - wreq = &pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("fo")}} + wreq = &pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("fo"), StartRevision: startRev}} } if err := wStream.Send(wreq); err != nil { t.Fatalf("wStream.Send error: %v", err) @@ -573,9 +581,16 @@ func TestV3WatchMultiple(t *testing.T) { clus.Terminate(t) } -// TestV3WatchMultipleEventsFromCurrentRevision tests Watch APIs from current revision -// in cases it receives multiple events. -func TestV3WatchMultipleEventsFromCurrentRevision(t *testing.T) { +func TestV3WatchMultipleEventsTxnSynced(t *testing.T) { + testV3WatchMultipleEventsTxn(t, 0) +} + +func TestV3WatchMultipleEventsTxnUnsynced(t *testing.T) { + testV3WatchMultipleEventsTxn(t, 1) +} + +// testV3WatchMultipleEventsTxn tests Watch APIs when it receives multiple events. +func testV3WatchMultipleEventsTxn(t *testing.T, startRev int64) { clus := newClusterGRPC(t, &clusterConfig{size: 3}) wAPI := pb.NewWatchClient(clus.RandConn()) @@ -584,7 +599,7 @@ func TestV3WatchMultipleEventsFromCurrentRevision(t *testing.T) { t.Fatalf("wAPI.Watch error: %v", wErr) } - if err := wStream.Send(&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("foo")}}); err != nil { + if err := wStream.Send(&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("foo"), StartRevision: startRev}}); err != nil { t.Fatalf("wStream.Send error: %v", err) } @@ -651,8 +666,16 @@ func (evs eventsSortByKey) Len() int { return len(evs) } func (evs eventsSortByKey) Swap(i, j int) { evs[i], evs[j] = evs[j], evs[i] } func (evs eventsSortByKey) Less(i, j int) bool { return bytes.Compare(evs[i].Kv.Key, evs[j].Kv.Key) < 0 } -// TestV3WatchMultipleStreams tests multiple watchers on the same key on multiple streams. -func TestV3WatchMultipleStreams(t *testing.T) { +func TestV3WatchMultipleStreamsSynced(t *testing.T) { + testV3WatchMultipleStreams(t, 0) +} + +func TestV3WatchMultipleStreamsUnsynced(t *testing.T) { + testV3WatchMultipleStreams(t, 1) +} + +// testV3WatchMultipleStreams tests multiple watchers on the same key on multiple streams. +func testV3WatchMultipleStreams(t *testing.T, startRev int64) { clus := newClusterGRPC(t, &clusterConfig{size: 3}) wAPI := pb.NewWatchClient(clus.RandConn()) kvc := pb.NewKVClient(clus.RandConn()) @@ -663,7 +686,7 @@ func TestV3WatchMultipleStreams(t *testing.T) { if errW != nil { t.Fatalf("wAPI.Watch error: %v", errW) } - if err := wStream.Send(&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo")}}); err != nil { + if err := wStream.Send(&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo"), StartRevision: startRev}}); err != nil { t.Fatalf("wStream.Send error: %v", err) } streams[i] = wStream