Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4214 from gyuho/watch_integration_test

Browse Source 
integration: add WatchFromCurrentRevision test
This commit is contained in:
Gyu-Ho Lee 2016-01-14 16:56:56 -08:00
commit cf873bcf81
1 changed files with 236 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

236
integration/v3_grpc_test.go
View File

@ -17,10 +17,12 @@ import (
"math/rand" "math/rand"
"reflect" "reflect"
"testing" "testing"
"time"
"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context" "github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
"github.com/coreos/etcd/Godeps/_workspace/src/google.golang.org/grpc" "github.com/coreos/etcd/Godeps/_workspace/src/google.golang.org/grpc"
pb "github.com/coreos/etcd/etcdserver/etcdserverpb" pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
"github.com/coreos/etcd/storage/storagepb"
) )
type clusterV3 struct { type clusterV3 struct {
@ -193,3 +195,237 @@ func TestV3DeleteRange(t *testing.T) {
clus.Terminate(t) clus.Terminate(t)
} }
} }
// TestV3WatchFromCurrentRevision tests Watch APIs from current revision.
func TestV3WatchFromCurrentRevision(t *testing.T) {
tests := []struct {
putKeys []string
watchRequest *pb.WatchRequest
wresps []*pb.WatchResponse
}{
// watch the key, matching
{
[]string{"foo"},
&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo")}},
[]*pb.WatchResponse{
{
Header: &pb.ResponseHeader{Revision: 1},
Created: true,
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
},
},
},
},
},
// watch the key, non-matching
{
[]string{"foo"},
&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Key: []byte("helloworld")}},
[]*pb.WatchResponse{
{
Header: &pb.ResponseHeader{Revision: 1},
Created: true,
},
},
},
// watch the prefix, matching
{
[]string{"fooLong"},
&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("foo")}},
[]*pb.WatchResponse{
{
Header: &pb.ResponseHeader{Revision: 1},
Created: true,
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("fooLong"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
},
},
},
},
},
// watch the prefix, non-matching
{
[]string{"foo"},
&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("helloworld")}},
[]*pb.WatchResponse{
{
Header: &pb.ResponseHeader{Revision: 1},
Created: true,
},
},
},
// multiple puts, one watcher with matching key
{
[]string{"foo", "foo", "foo"},
&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo")}},
[]*pb.WatchResponse{
{
Header: &pb.ResponseHeader{Revision: 1},
Created: true,
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
},
},
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 2, Version: 2},
},
},
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 3, Version: 3},
},
},
},
},
},
// multiple puts, one watcher with matching prefix
{
[]string{"foo", "foo", "foo"},
&pb.WatchRequest{CreateRequest: &pb.WatchCreateRequest{Prefix: []byte("foo")}},
[]*pb.WatchResponse{
{
Header: &pb.ResponseHeader{Revision: 1},
Created: true,
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 1, Version: 1},
},
},
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 2, Version: 2},
},
},
},
{
Header: &pb.ResponseHeader{Revision: 1},
Created: false,
Events: []*storagepb.Event{
{
Type: storagepb.PUT,
Kv: &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 1, ModRevision: 3, Version: 3},
},
},
},
},
},
// TODO: watch and receive multiple-events from synced (need Txn)
}
for i, tt := range tests {
clus := newClusterGRPC(t, &clusterConfig{size: 3})
wAPI := pb.NewWatchClient(clus.RandConn())
wStream, err := wAPI.Watch(context.TODO())
if err != nil {
t.Fatalf("#%d: wAPI.Watch error: %v", i, err)
}
if err := wStream.Send(tt.watchRequest); err != nil {
t.Fatalf("#%d: wStream.Send error: %v", i, err)
}
kvc := pb.NewKVClient(clus.RandConn())
for _, k := range tt.putKeys {
if _, err := kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte(k), Value: []byte("bar")}); err != nil {
t.Fatalf("#%d: couldn't put key (%v)", i, err)
}
}
var createdWatchId int64
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.Logf("[TODO - skip for now] #%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.Created {
createdWatchId = resp.WatchId
}
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)
}
}
rCh := make(chan *pb.WatchResponse)
go func() {
resp, _ := wStream.Recv()
rCh <- resp
}()
select {
case nr := <-rCh:
t.Errorf("#%d: unexpected response is received %+v", i, nr)
case <-time.After(2 * time.Second):
}
wStream.CloseSend()
rv, ok := <-rCh
if rv != nil || !ok {
t.Errorf("#%d: rv, ok got = %v %v, want = nil true", i, rv, ok)
}
// can't defer because tcp ports will be in use
clus.Terminate(t)
}
}