Function TestWatchRequestProgress

tests/integration/clientv3/watch_test.go:611–677

(t *testing.T)

}

func TestWatchRequestProgress(t *testing.T) {
	if integration2.ThroughProxy {
		t.Skipf("grpc-proxy does not support WatchProgress yet")
	}
	testCases := []struct {
		name     string
		watchers []string
	}{
		{"0-watcher", []string{}},
		{"1-watcher", []string{"/"}},
		{"2-watcher", []string{"/", "/"}},
	}

	for _, c := range testCases {
		t.Run(c.name, func(t *testing.T) {
			integration2.BeforeTest(t)

			watchTimeout := 3 * time.Second

			clus := integration2.NewCluster(t, &integration2.ClusterConfig{Size: 3})
			defer clus.Terminate(t)

			wc := clus.RandClient()

			var watchChans []clientv3.WatchChan

			for _, prefix := range c.watchers {
				watchChans = append(watchChans, wc.Watch(context.Background(), prefix, clientv3.WithPrefix()))
			}

			_, err := wc.Put(context.Background(), "/a", "1")
			require.NoError(t, err)

			for _, rch := range watchChans {
				select {
				case resp := <-rch: // wait for notification
					if len(resp.Events) != 1 {
						t.Fatalf("resp.Events expected 1, got %d", len(resp.Events))
					}
				case <-time.After(watchTimeout):
					t.Fatalf("watch response expected in %v, but timed out", watchTimeout)
				}
			}

			// put a value not being watched to increment revision
			_, err = wc.Put(context.Background(), "x", "1")
			require.NoError(t, err)

			require.NoError(t, wc.RequestProgress(context.Background()))

			// verify all watch channels receive a progress notify
			for _, rch := range watchChans {
				select {
				case resp := <-rch:
					if !resp.IsProgressNotify() {
						t.Fatalf("expected resp.IsProgressNotify() == true")
					}
					if resp.Header.Revision != 3 {