mirror_test.go 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. // Copyright 2016 CoreOS, Inc.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package integration
  15. import (
  16. "reflect"
  17. "testing"
  18. "time"
  19. "github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
  20. "github.com/coreos/etcd/clientv3/mirror"
  21. "github.com/coreos/etcd/integration"
  22. "github.com/coreos/etcd/pkg/testutil"
  23. "github.com/coreos/etcd/storage/storagepb"
  24. )
  25. func TestMirrorSync(t *testing.T) {
  26. defer testutil.AfterTest(t)
  27. clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
  28. defer clus.Terminate(t)
  29. c := clus.Client(0)
  30. _, err := c.KV.Put(context.TODO(), "foo", "bar")
  31. if err != nil {
  32. t.Fatal(err)
  33. }
  34. syncer := mirror.NewSyncer(c, "", 0)
  35. gch, ech := syncer.SyncBase(context.TODO())
  36. wkvs := []*storagepb.KeyValue{{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 2, Version: 1}}
  37. for g := range gch {
  38. if !reflect.DeepEqual(g.Kvs, wkvs) {
  39. t.Fatalf("kv = %v, want %v", g.Kvs, wkvs)
  40. }
  41. }
  42. for e := range ech {
  43. t.Fatalf("unexpected error %v", e)
  44. }
  45. wch := syncer.SyncUpdates(context.TODO())
  46. _, err = c.KV.Put(context.TODO(), "foo", "bar")
  47. if err != nil {
  48. t.Fatal(err)
  49. }
  50. select {
  51. case r := <-wch:
  52. wkv := &storagepb.KeyValue{Key: []byte("foo"), Value: []byte("bar"), CreateRevision: 2, ModRevision: 3, Version: 2}
  53. if !reflect.DeepEqual(r.Events[0].Kv, wkv) {
  54. t.Fatalf("kv = %v, want %v", r.Events[0].Kv, wkv)
  55. }
  56. case <-time.After(time.Second):
  57. t.Fatal("failed to receive update in one second")
  58. }
  59. }