integration: some leasing tests

clientv3/integration/leasing_test.go

@@ -15,9 +15,11 @@ package integration
 
 import (
 	"context"
+	"reflect"
 	"testing"
 
-	leasing "github.com/coreos/etcd/clientv3/leasing"
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/clientv3/leasing"
 	"github.com/coreos/etcd/integration"
 	"github.com/coreos/etcd/pkg/testutil"
 )
@@ -117,3 +119,200 @@ func TestLeasingGet2(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+// TestLeasingInterval checks the leasing KV fetches key intervals.
+func TestLeasingInterval(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	lkv, err := leasing.NewleasingKV(clus.Client(0), "pfx/")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	keys := []string{"abc/a", "abc/b", "abc/c"}
+	for _, k := range keys {
+		if _, err := clus.Client(0).Put(context.TODO(), k, "v"); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	resp, err := lkv.Get(context.TODO(), "abc/", clientv3.WithPrefix())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(resp.Kvs) != 3 {
+		t.Fatalf("expected keys %+v, got response keys %+v", keys, resp.Kvs)
+	}
+}
+
+// TestLeasingPutInvalidateNew checks the leasing KV updates its cache on a Put to a new key.
+func TestLeasingPutInvalidateNew(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	lkv, err := leasing.NewleasingKV(clus.Client(0), "pfx/")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := lkv.Get(context.TODO(), "k"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := lkv.Put(context.TODO(), "k", "v"); err != nil {
+		t.Fatal(err)
+	}
+
+	lkvResp, err := lkv.Get(context.TODO(), "k")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cResp, cerr := clus.Client(0).Get(context.TODO(), "k")
+	if cerr != nil {
+		t.Fatal(cerr)
+	}
+	if !reflect.DeepEqual(lkvResp, cResp) {
+		t.Fatalf(`expected %+v, got response %+v`, cResp, lkvResp)
+	}
+}
+
+// TestLeasingPutInvalidateExisting checks the leasing KV updates its cache on a Put to an existing key.
+func TestLeasingPutInvalidatExisting(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	if _, err := clus.Client(0).Put(context.TODO(), "k", "abc"); err != nil {
+		t.Fatal(err)
+	}
+
+	lkv, err := leasing.NewleasingKV(clus.Client(0), "pfx/")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := lkv.Get(context.TODO(), "k"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := lkv.Put(context.TODO(), "k", "v"); err != nil {
+		t.Fatal(err)
+	}
+
+	lkvResp, err := lkv.Get(context.TODO(), "k")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cResp, cerr := clus.Client(0).Get(context.TODO(), "k")
+	if cerr != nil {
+		t.Fatal(cerr)
+	}
+	if !reflect.DeepEqual(lkvResp, cResp) {
+		t.Fatalf(`expected %+v, got response %+v`, cResp, lkvResp)
+	}
+}
+
+// TestLeasingGetSerializable checks the leasing KV can make serialized requests
+// when the etcd cluster is partitioned.
+func TestLeasingGetSerializable(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 2})
+	defer clus.Terminate(t)
+
+	lkv, err := leasing.NewleasingKV(clus.Client(0), "pfx/")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := clus.Client(0).Put(context.TODO(), "cached", "abc"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := lkv.Get(context.TODO(), "cached"); err != nil {
+		t.Fatal(err)
+	}
+
+	clus.Members[1].Stop(t)
+
+	// leasing key ownership should have "cached" locally served
+	cachedResp, err := lkv.Get(context.TODO(), "cached", clientv3.WithSerializable())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(cachedResp.Kvs) != 1 || string(cachedResp.Kvs[0].Value) != "abc" {
+		t.Fatalf(`expected "cached"->"abc", got response %+v`, cachedResp)
+	}
+
+	// don't necessarily try to acquire leasing key ownership for new key
+	resp, err := lkv.Get(context.TODO(), "uncached", clientv3.WithSerializable())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(resp.Kvs) != 0 {
+		t.Fatalf(`expected no keys, got response %+v`, resp)
+	}
+}
+
+// TestLeasingPrevKey checks the cache respects the PrevKV flag on puts.
+func TestLeasingPrevKey(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 2})
+	defer clus.Terminate(t)
+
+	lkv, err := leasing.NewleasingKV(clus.Client(0), "pfx/")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := clus.Client(0).Put(context.TODO(), "k", "abc"); err != nil {
+		t.Fatal(err)
+	}
+	// fetch without prevkv to acquire leasing key
+	if _, err := lkv.Get(context.TODO(), "k"); err != nil {
+		t.Fatal(err)
+	}
+	// fetch prevkv via put
+	resp, err := lkv.Put(context.TODO(), "k", "def", clientv3.WithPrevKV())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.PrevKv == nil || string(resp.PrevKv.Value) != "abc" {
+		t.Fatalf(`expected PrevKV.Value="abc", got response %+v`, resp)
+	}
+}
+
+// TestLeasingRevGet checks the cache respects Get by Revision.
+func TestLeasingRevGet(t *testing.T) {
+	defer testutil.AfterTest(t)
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	lkv, err := leasing.NewleasingKV(clus.Client(0), "pfx/")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	putResp, err := clus.Client(0).Put(context.TODO(), "k", "abc")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := clus.Client(0).Put(context.TODO(), "k", "def"); err != nil {
+		t.Fatal(err)
+	}
+
+	// check historic revision
+	getResp, gerr := lkv.Get(context.TODO(), "k", clientv3.WithRev(putResp.Header.Revision))
+	if gerr != nil {
+		t.Fatal(gerr)
+	}
+	if len(getResp.Kvs) != 1 || string(getResp.Kvs[0].Value) != "abc" {
+		t.Fatalf(`expeted "k"->"abc" at rev=%d, got response %+v`, putResp.Header.Revision, getResp)
+	}
+	// check current revision
+	getResp, gerr = lkv.Get(context.TODO(), "k")
+	if gerr != nil {
+		t.Fatal(gerr)
+	}
+	if len(getResp.Kvs) != 1 || string(getResp.Kvs[0].Value) != "def" {
+		t.Fatalf(`expeted "k"->"abc" at rev=%d, got response %+v`, putResp.Header.Revision, getResp)
+	}
+}