etcd/client/client_test.go

// Copyright 2015 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package client

import (
	"context"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net/http"
	"net/url"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	"go.etcd.io/etcd/pkg/testutil"
	"go.etcd.io/etcd/version"
)

type actionAssertingHTTPClient struct {
	t   *testing.T
	num int
	act httpAction

	resp http.Response
	body []byte
	err  error
}

func (a *actionAssertingHTTPClient) Do(_ context.Context, act httpAction) (*http.Response, []byte, error) {
	if !reflect.DeepEqual(a.act, act) {
		a.t.Errorf("#%d: unexpected httpAction: want=%#v got=%#v", a.num, a.act, act)
	}

	return &a.resp, a.body, a.err
}

type staticHTTPClient struct {
	resp http.Response
	body []byte
	err  error
}

func (s *staticHTTPClient) Do(context.Context, httpAction) (*http.Response, []byte, error) {
	return &s.resp, s.body, s.err
}

type staticHTTPAction struct {
	request http.Request
}

func (s *staticHTTPAction) HTTPRequest(url.URL) *http.Request {
	return &s.request
}

type staticHTTPResponse struct {
	resp http.Response
	body []byte
	err  error
}

type multiStaticHTTPClient struct {
	responses []staticHTTPResponse
	cur       int
}

func (s *multiStaticHTTPClient) Do(context.Context, httpAction) (*http.Response, []byte, error) {
	r := s.responses[s.cur]
	s.cur++
	return &r.resp, r.body, r.err
}

func newStaticHTTPClientFactory(responses []staticHTTPResponse) httpClientFactory {
	var cur int
	return func(url.URL) httpClient {
		r := responses[cur]
		cur++
		return &staticHTTPClient{resp: r.resp, body: r.body, err: r.err}
	}
}

type fakeTransport struct {
	respchan     chan *http.Response
	errchan      chan error
	startCancel  chan struct{}
	finishCancel chan struct{}
}

func newFakeTransport() *fakeTransport {
	return &fakeTransport{
		respchan:     make(chan *http.Response, 1),
		errchan:      make(chan error, 1),
		startCancel:  make(chan struct{}, 1),
		finishCancel: make(chan struct{}, 1),
	}
}

func (t *fakeTransport) CancelRequest(*http.Request) {
	t.startCancel <- struct{}{}
}

type fakeAction struct{}

func (a *fakeAction) HTTPRequest(url.URL) *http.Request {
	return &http.Request{}
}

func TestSimpleHTTPClientDoSuccess(t *testing.T) {
	tr := newFakeTransport()
	c := &simpleHTTPClient{transport: tr}

	tr.respchan <- &http.Response{
		StatusCode: http.StatusTeapot,
		Body:       ioutil.NopCloser(strings.NewReader("foo")),
	}

	resp, body, err := c.Do(context.Background(), &fakeAction{})
	if err != nil {
		t.Fatalf("incorrect error value: want=nil got=%v", err)
	}

	wantCode := http.StatusTeapot
	if wantCode != resp.StatusCode {
		t.Fatalf("invalid response code: want=%d got=%d", wantCode, resp.StatusCode)
	}

	wantBody := []byte("foo")
	if !reflect.DeepEqual(wantBody, body) {
		t.Fatalf("invalid response body: want=%q got=%q", wantBody, body)
	}
}

func TestSimpleHTTPClientDoError(t *testing.T) {
	tr := newFakeTransport()
	c := &simpleHTTPClient{transport: tr}

	tr.errchan <- errors.New("fixture")

	_, _, err := c.Do(context.Background(), &fakeAction{})
	if err == nil {
		t.Fatalf("expected non-nil error, got nil")
	}
}

func TestSimpleHTTPClientDoCancelContext(t *testing.T) {
	tr := newFakeTransport()
	c := &simpleHTTPClient{transport: tr}

	tr.startCancel <- struct{}{}
	tr.finishCancel <- struct{}{}

	_, _, err := c.Do(context.Background(), &fakeAction{})
	if err == nil {
		t.Fatalf("expected non-nil error, got nil")
	}
}

type checkableReadCloser struct {
	io.ReadCloser
	closed bool
}

func (c *checkableReadCloser) Close() error {
	if !c.closed {
		c.closed = true
		return c.ReadCloser.Close()
	}
	return nil
}

func TestSimpleHTTPClientDoCancelContextResponseBodyClosed(t *testing.T) {
	tr := newFakeTransport()
	c := &simpleHTTPClient{transport: tr}

	// create an already-cancelled context
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	body := &checkableReadCloser{ReadCloser: ioutil.NopCloser(strings.NewReader("foo"))}
	go func() {
		// wait that simpleHTTPClient knows the context is already timed out,
		// and calls CancelRequest
		testutil.WaitSchedule()

		// response is returned before cancel effects
		tr.respchan <- &http.Response{Body: body}
	}()

	_, _, err := c.Do(ctx, &fakeAction{})
	if err == nil {
		t.Fatalf("expected non-nil error, got nil")
	}

	if !body.closed {
		t.Fatalf("expected closed body")
	}
}

type blockingBody struct {
	c chan struct{}
}

func (bb *blockingBody) Read(p []byte) (n int, err error) {
	<-bb.c
	return 0, errors.New("closed")
}

func (bb *blockingBody) Close() error {
	close(bb.c)
	return nil
}

func TestSimpleHTTPClientDoCancelContextResponseBodyClosedWithBlockingBody(t *testing.T) {
	tr := newFakeTransport()
	c := &simpleHTTPClient{transport: tr}

	ctx, cancel := context.WithCancel(context.Background())
	body := &checkableReadCloser{ReadCloser: &blockingBody{c: make(chan struct{})}}
	go func() {
		tr.respchan <- &http.Response{Body: body}
		time.Sleep(2 * time.Millisecond)
		// cancel after the body is received
		cancel()
	}()

	_, _, err := c.Do(ctx, &fakeAction{})
	if err != context.Canceled {
		t.Fatalf("expected %+v, got %+v", context.Canceled, err)
	}

	if !body.closed {
		t.Fatalf("expected closed body")
	}
}

func TestSimpleHTTPClientDoCancelContextWaitForRoundTrip(t *testing.T) {
	tr := newFakeTransport()
	c := &simpleHTTPClient{transport: tr}

	donechan := make(chan struct{})
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		c.Do(ctx, &fakeAction{})
		close(donechan)
	}()

	// This should call CancelRequest and begin the cancellation process
	cancel()

	select {
	case <-donechan:
		t.Fatalf("simpleHTTPClient.Do should not have exited yet")
	default:
	}

	tr.finishCancel <- struct{}{}

	select {
	case <-donechan:
		//expected behavior
		return
	case <-time.After(time.Second):
		t.Fatalf("simpleHTTPClient.Do did not exit within 1s")
	}
}

func TestSimpleHTTPClientDoHeaderTimeout(t *testing.T) {
	tr := newFakeTransport()
	tr.finishCancel <- struct{}{}
	c := &simpleHTTPClient{transport: tr, headerTimeout: time.Millisecond}

	errc := make(chan error)
	go func() {
		_, _, err := c.Do(context.Background(), &fakeAction{})
		errc <- err
	}()

	select {
	case err := <-errc:
		if err == nil {
			t.Fatalf("expected non-nil error, got nil")
		}
	case <-time.After(time.Second):
		t.Fatalf("unexpected timeout when waiting for the test to finish")
	}
}

func TestHTTPClusterClientDo(t *testing.T) {
	fakeErr := errors.New("fake!")
	fakeURL := url.URL{}
	tests := []struct {
		client *httpClusterClient
		ctx    context.Context

		wantCode   int
		wantErr    error
		wantPinned int
	}{
		// first good response short-circuits Do
		{
			client: &httpClusterClient{
				endpoints: []url.URL{fakeURL, fakeURL},
				clientFactory: newStaticHTTPClientFactory(
					[]staticHTTPResponse{
						{resp: http.Response{StatusCode: http.StatusTeapot}},
						{err: fakeErr},
					},
				),
				rand: rand.New(rand.NewSource(0)),
			},
			wantCode: http.StatusTeapot,
		},

		// fall through to good endpoint if err is arbitrary
		{
			client: &httpClusterClient{
				endpoints: []url.URL{fakeURL, fakeURL},
				clientFactory: newStaticHTTPClientFactory(
					[]staticHTTPResponse{
						{err: fakeErr},
						{resp: http.Response{StatusCode: http.StatusTeapot}},
					},
				),
				rand: rand.New(rand.NewSource(0)),
			},
			wantCode:   http.StatusTeapot,
			wantPinned: 1,
		},

		// context.Canceled short-circuits Do
		{
			client: &httpClusterClient{
				endpoints: []url.URL{fakeURL, fakeURL},
				clientFactory: newStaticHTTPClientFactory(
					[]staticHTTPResponse{
						{err: context.Canceled},
						{resp: http.Response{StatusCode: http.StatusTeapot}},
					},
				),
				rand: rand.New(rand.NewSource(0)),
			},
			wantErr: context.Canceled,
		},

		// return err if there are no endpoints
		{
			client: &httpClusterClient{
				endpoints:     []url.URL{},
				clientFactory: newHTTPClientFactory(nil, nil, 0),
				rand:          rand.New(rand.NewSource(0)),
			},
			wantErr: ErrNoEndpoints,
		},

		// return err if all endpoints return arbitrary errors
		{
			client: &httpClusterClient{
				endpoints: []url.URL{fakeURL, fakeURL},
				clientFactory: newStaticHTTPClientFactory(
					[]staticHTTPResponse{
						{err: fakeErr},
						{err: fakeErr},
					},
				),
				rand: rand.New(rand.NewSource(0)),
			},
			wantErr: &ClusterError{Errors: []error{fakeErr, fakeErr}},
		},

		// 500-level errors cause Do to fallthrough to next endpoint
		{
			client: &httpClusterClient{
				endpoints: []url.URL{fakeURL, fakeURL},
				clientFactory: newStaticHTTPClientFactory(
					[]staticHTTPResponse{
						{resp: http.Response{StatusCode: http.StatusBadGateway}},
						{resp: http.Response{StatusCode: http.StatusTeapot}},
					},
				),
				rand: rand.New(rand.NewSource(0)),
			},
			wantCode:   http.StatusTeapot,
			wantPinned: 1,
		},

		// 500-level errors cause one shot Do to fallthrough to next endpoint
		{
			client: &httpClusterClient{
				endpoints: []url.URL{fakeURL, fakeURL},
				clientFactory: newStaticHTTPClientFactory(
					[]staticHTTPResponse{
						{resp: http.Response{StatusCode: http.StatusBadGateway}},
						{resp: http.Response{StatusCode: http.StatusTeapot}},
					},
				),
				rand: rand.New(rand.NewSource(0)),
			},
			ctx:        context.WithValue(context.Background(), &oneShotCtxValue, &oneShotCtxValue),
			wantErr:    fmt.Errorf("client: etcd member  returns server error [Bad Gateway]"),
			wantPinned: 1,
		},
	}

	for i, tt := range tests {
		if tt.ctx == nil {
			tt.ctx = context.Background()
		}
		resp, _, err := tt.client.Do(tt.ctx, nil)
		if (tt.wantErr == nil && tt.wantErr != err) || (tt.wantErr != nil && tt.wantErr.Error() != err.Error()) {
			t.Errorf("#%d: got err=%v, want=%v", i, err, tt.wantErr)
			continue
		}

		if resp == nil {
			if tt.wantCode != 0 {
				t.Errorf("#%d: resp is nil, want=%d", i, tt.wantCode)
				continue
			}
		} else if resp.StatusCode != tt.wantCode {
			t.Errorf("#%d: resp code=%d, want=%d", i, resp.StatusCode, tt.wantCode)
			continue
		}

		if tt.client.pinned != tt.wantPinned {
			t.Errorf("#%d: pinned=%d, want=%d", i, tt.client.pinned, tt.wantPinned)
		}
	}
}

func TestHTTPClusterClientDoDeadlineExceedContext(t *testing.T) {
	fakeURL := url.URL{}
	tr := newFakeTransport()
	tr.finishCancel <- struct{}{}
	c := &httpClusterClient{
		clientFactory: newHTTPClientFactory(tr, DefaultCheckRedirect, 0),
		endpoints:     []url.URL{fakeURL},
	}

	errc := make(chan error)
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond)
		defer cancel()
		_, _, err := c.Do(ctx, &fakeAction{})
		errc <- err
	}()

	select {
	case err := <-errc:
		if err != context.DeadlineExceeded {
			t.Errorf("err = %+v, want %+v", err, context.DeadlineExceeded)
		}
	case <-time.After(time.Second):
		t.Fatalf("unexpected timeout when waiting for request to deadline exceed")
	}
}

type fakeCancelContext struct{}

var errFakeCancelContext = errors.New("fake context canceled")

func (f fakeCancelContext) Deadline() (time.Time, bool) { return time.Time{}, false }
func (f fakeCancelContext) Done() <-chan struct{} {
	d := make(chan struct{}, 1)
	d <- struct{}{}
	return d
}
func (f fakeCancelContext) Err() error                        { return errFakeCancelContext }
func (f fakeCancelContext) Value(key interface{}) interface{} { return 1 }

func withTimeout(parent context.Context, timeout time.Duration) (
	ctx context.Context,
	cancel context.CancelFunc) {
	ctx = parent
	cancel = func() {
		ctx = nil
	}
	return ctx, cancel
}

func TestHTTPClusterClientDoCanceledContext(t *testing.T) {
	fakeURL := url.URL{}
	tr := newFakeTransport()
	tr.finishCancel <- struct{}{}
	c := &httpClusterClient{
		clientFactory: newHTTPClientFactory(tr, DefaultCheckRedirect, 0),
		endpoints:     []url.URL{fakeURL},
	}

	errc := make(chan error)
	go func() {
		ctx, cancel := withTimeout(fakeCancelContext{}, time.Millisecond)
		cancel()
		_, _, err := c.Do(ctx, &fakeAction{})
		errc <- err
	}()

	select {
	case err := <-errc:
		if err != errFakeCancelContext {
			t.Errorf("err = %+v, want %+v", err, errFakeCancelContext)
		}
	case <-time.After(time.Second):
		t.Fatalf("unexpected timeout when waiting for request to fake context canceled")
	}
}

func TestRedirectedHTTPAction(t *testing.T) {
	act := &redirectedHTTPAction{
		action: &staticHTTPAction{
			request: http.Request{
				Method: "DELETE",
				URL: &url.URL{
					Scheme: "https",
					Host:   "foo.example.com",
					Path:   "/ping",
				},
			},
		},
		location: url.URL{
			Scheme: "https",
			Host:   "bar.example.com",
			Path:   "/pong",
		},
	}

	want := &http.Request{
		Method: "DELETE",
		URL: &url.URL{
			Scheme: "https",
			Host:   "bar.example.com",
			Path:   "/pong",
		},
	}
	got := act.HTTPRequest(url.URL{Scheme: "http", Host: "baz.example.com", Path: "/pang"})

	if !reflect.DeepEqual(want, got) {
		t.Fatalf("HTTPRequest is %#v, want %#v", want, got)
	}
}

func TestRedirectFollowingHTTPClient(t *testing.T) {
	tests := []struct {
		checkRedirect CheckRedirectFunc
		client        httpClient
		wantCode      int
		wantErr       error
	}{
		// errors bubbled up
		{
			checkRedirect: func(int) error { return ErrTooManyRedirects },
			client: &multiStaticHTTPClient{
				responses: []staticHTTPResponse{
					{
						err: errors.New("fail!"),
					},
				},
			},
			wantErr: errors.New("fail!"),
		},

		// no need to follow redirect if none given
		{
			checkRedirect: func(int) error { return ErrTooManyRedirects },
			client: &multiStaticHTTPClient{
				responses: []staticHTTPResponse{
					{
						resp: http.Response{
							StatusCode: http.StatusTeapot,
						},
					},
				},
			},
			wantCode: http.StatusTeapot,
		},

		// redirects if less than max
		{
			checkRedirect: func(via int) error {
				if via >= 2 {
					return ErrTooManyRedirects
				}
				return nil
			},
			client: &multiStaticHTTPClient{
				responses: []staticHTTPResponse{
					{
						resp: http.Response{
							StatusCode: http.StatusTemporaryRedirect,
							Header:     http.Header{"Location": []string{"http://example.com"}},
						},
					},
					{
						resp: http.Response{
							StatusCode: http.StatusTeapot,
						},
					},
				},
			},
			wantCode: http.StatusTeapot,
		},

		// succeed after reaching max redirects
		{
			checkRedirect: func(via int) error {
				if via >= 3 {
					return ErrTooManyRedirects
				}
				return nil
			},
			client: &multiStaticHTTPClient{
				responses: []staticHTTPResponse{
					{
						resp: http.Response{
							StatusCode: http.StatusTemporaryRedirect,
							Header:     http.Header{"Location": []string{"http://example.com"}},
						},
					},
					{
						resp: http.Response{
							StatusCode: http.StatusTemporaryRedirect,
							Header:     http.Header{"Location": []string{"http://example.com"}},
						},
					},
					{
						resp: http.Response{
							StatusCode: http.StatusTeapot,
						},
					},
				},
			},
			wantCode: http.StatusTeapot,
		},

		// fail if too many redirects
		{
			checkRedirect: func(via int) error {
				if via >= 2 {
					return ErrTooManyRedirects
				}
				return nil
			},
			client: &multiStaticHTTPClient{
				responses: []staticHTTPResponse{
					{
						resp: http.Response{
							StatusCode: http.StatusTemporaryRedirect,
							Header:     http.Header{"Location": []string{"http://example.com"}},
						},
					},
					{
						resp: http.Response{
							StatusCode: http.StatusTemporaryRedirect,
							Header:     http.Header{"Location": []string{"http://example.com"}},
						},
					},
					{
						resp: http.Response{
							StatusCode: http.StatusTeapot,
						},
					},
				},
			},
			wantErr: ErrTooManyRedirects,
		},

		// fail if Location header not set
		{
			checkRedirect: func(int) error { return ErrTooManyRedirects },
			client: &multiStaticHTTPClient{
				responses: []staticHTTPResponse{
					{
						resp: http.Response{
							StatusCode: http.StatusTemporaryRedirect,
						},
					},
				},
			},
			wantErr: errors.New("location header not set"),
		},

		// fail if Location header is invalid
		{
			checkRedirect: func(int) error { return ErrTooManyRedirects },
			client: &multiStaticHTTPClient{
				responses: []staticHTTPResponse{
					{
						resp: http.Response{
							StatusCode: http.StatusTemporaryRedirect,
							Header:     http.Header{"Location": []string{":"}},
						},
					},
				},
			},
			wantErr: errors.New("location header not valid URL: :"),
		},

		// fail if redirects checked way too many times
		{
			checkRedirect: func(int) error { return nil },
			client: &staticHTTPClient{
				resp: http.Response{
					StatusCode: http.StatusTemporaryRedirect,
					Header:     http.Header{"Location": []string{"http://example.com"}},
				},
			},
			wantErr: errTooManyRedirectChecks,
		},
	}

	for i, tt := range tests {
		client := &redirectFollowingHTTPClient{client: tt.client, checkRedirect: tt.checkRedirect}
		resp, _, err := client.Do(context.Background(), nil)
		if (tt.wantErr == nil && tt.wantErr != err) || (tt.wantErr != nil && tt.wantErr.Error() != err.Error()) {
			t.Errorf("#%d: got err=%v, want=%v", i, err, tt.wantErr)
			continue
		}

		if resp == nil {
			if tt.wantCode != 0 {
				t.Errorf("#%d: resp is nil, want=%d", i, tt.wantCode)
			}
			continue
		}

		if resp.StatusCode != tt.wantCode {
			t.Errorf("#%d: resp code=%d, want=%d", i, resp.StatusCode, tt.wantCode)
			continue
		}
	}
}

func TestDefaultCheckRedirect(t *testing.T) {
	tests := []struct {
		num int
		err error
	}{
		{0, nil},
		{5, nil},
		{10, nil},
		{11, ErrTooManyRedirects},
		{29, ErrTooManyRedirects},
	}

	for i, tt := range tests {
		err := DefaultCheckRedirect(tt.num)
		if !reflect.DeepEqual(tt.err, err) {
			t.Errorf("#%d: want=%#v got=%#v", i, tt.err, err)
		}
	}
}

func TestHTTPClusterClientSync(t *testing.T) {
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
	}
	err := hc.SetEndpoints([]string{"http://127.0.0.1:2379"})
	if err != nil {
		t.Fatalf("unexpected error during setup: %#v", err)
	}

	want := []string{"http://127.0.0.1:2379"}
	got := hc.Endpoints()
	if !reflect.DeepEqual(want, got) {
		t.Fatalf("incorrect endpoints: want=%#v got=%#v", want, got)
	}

	err = hc.Sync(context.Background())
	if err != nil {
		t.Fatalf("unexpected error during Sync: %#v", err)
	}

	want = []string{"http://127.0.0.1:2379", "http://127.0.0.1:4001", "http://127.0.0.1:4002", "http://127.0.0.1:4003"}
	got = hc.Endpoints()
	sort.Strings(got)
	if !reflect.DeepEqual(want, got) {
		t.Fatalf("incorrect endpoints post-Sync: want=%#v got=%#v", want, got)
	}

	err = hc.SetEndpoints([]string{"http://127.0.0.1:4009"})
	if err != nil {
		t.Fatalf("unexpected error during reset: %#v", err)
	}

	want = []string{"http://127.0.0.1:4009"}
	got = hc.Endpoints()
	if !reflect.DeepEqual(want, got) {
		t.Fatalf("incorrect endpoints post-reset: want=%#v got=%#v", want, got)
	}
}

func TestHTTPClusterClientSyncFail(t *testing.T) {
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{err: errors.New("fail!")},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
	}
	err := hc.SetEndpoints([]string{"http://127.0.0.1:2379"})
	if err != nil {
		t.Fatalf("unexpected error during setup: %#v", err)
	}

	want := []string{"http://127.0.0.1:2379"}
	got := hc.Endpoints()
	if !reflect.DeepEqual(want, got) {
		t.Fatalf("incorrect endpoints: want=%#v got=%#v", want, got)
	}

	err = hc.Sync(context.Background())
	if err == nil {
		t.Fatalf("got nil error during Sync")
	}

	got = hc.Endpoints()
	if !reflect.DeepEqual(want, got) {
		t.Fatalf("incorrect endpoints after failed Sync: want=%#v got=%#v", want, got)
	}
}

func TestHTTPClusterClientAutoSyncCancelContext(t *testing.T) {
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
	}
	err := hc.SetEndpoints([]string{"http://127.0.0.1:2379"})
	if err != nil {
		t.Fatalf("unexpected error during setup: %#v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	err = hc.AutoSync(ctx, time.Hour)
	if err != context.Canceled {
		t.Fatalf("incorrect error value: want=%v got=%v", context.Canceled, err)
	}
}

func TestHTTPClusterClientAutoSyncFail(t *testing.T) {
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{err: errors.New("fail!")},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
	}
	err := hc.SetEndpoints([]string{"http://127.0.0.1:2379"})
	if err != nil {
		t.Fatalf("unexpected error during setup: %#v", err)
	}

	err = hc.AutoSync(context.Background(), time.Hour)
	if !strings.HasPrefix(err.Error(), ErrClusterUnavailable.Error()) {
		t.Fatalf("incorrect error value: want=%v got=%v", ErrClusterUnavailable, err)
	}
}

func TestHTTPClusterClientGetVersion(t *testing.T) {
	body := []byte(`{"etcdserver":"2.3.2","etcdcluster":"2.3.0"}`)
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Length": []string{"44"}}},
			body: body,
		},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
	}
	err := hc.SetEndpoints([]string{"http://127.0.0.1:4003", "http://127.0.0.1:2379", "http://127.0.0.1:4001", "http://127.0.0.1:4002"})
	if err != nil {
		t.Fatalf("unexpected error during setup: %#v", err)
	}

	actual, err := hc.GetVersion(context.Background())
	if err != nil {
		t.Errorf("non-nil error: %#v", err)
	}
	expected := version.Versions{Server: "2.3.2", Cluster: "2.3.0"}
	if !reflect.DeepEqual(&expected, actual) {
		t.Errorf("incorrect Response: want=%#v got=%#v", expected, actual)
	}
}

// TestHTTPClusterClientSyncPinEndpoint tests that Sync() pins the endpoint when
// it gets the exactly same member list as before.
func TestHTTPClusterClientSyncPinEndpoint(t *testing.T) {
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
	}
	err := hc.SetEndpoints([]string{"http://127.0.0.1:4003", "http://127.0.0.1:2379", "http://127.0.0.1:4001", "http://127.0.0.1:4002"})
	if err != nil {
		t.Fatalf("unexpected error during setup: %#v", err)
	}
	pinnedEndpoint := hc.endpoints[hc.pinned]

	for i := 0; i < 3; i++ {
		err = hc.Sync(context.Background())
		if err != nil {
			t.Fatalf("#%d: unexpected error during Sync: %#v", i, err)
		}

		if g := hc.endpoints[hc.pinned]; g != pinnedEndpoint {
			t.Errorf("#%d: pinned endpoint = %v, want %v", i, g, pinnedEndpoint)
		}
	}
}

// TestHTTPClusterClientSyncUnpinEndpoint tests that Sync() unpins the endpoint when
// it gets a different member list than before.
func TestHTTPClusterClientSyncUnpinEndpoint(t *testing.T) {
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
	}
	err := hc.SetEndpoints([]string{"http://127.0.0.1:4003", "http://127.0.0.1:2379", "http://127.0.0.1:4001", "http://127.0.0.1:4002"})
	if err != nil {
		t.Fatalf("unexpected error during setup: %#v", err)
	}
	wants := []string{"http://127.0.0.1:2379", "http://127.0.0.1:4001", "http://127.0.0.1:4002"}

	for i := 0; i < 3; i++ {
		err = hc.Sync(context.Background())
		if err != nil {
			t.Fatalf("#%d: unexpected error during Sync: %#v", i, err)
		}

		if g := hc.endpoints[hc.pinned]; g.String() != wants[i] {
			t.Errorf("#%d: pinned endpoint = %v, want %v", i, g, wants[i])
		}
	}
}

// TestHTTPClusterClientSyncPinLeaderEndpoint tests that Sync() pins the leader
// when the selection mode is EndpointSelectionPrioritizeLeader
func TestHTTPClusterClientSyncPinLeaderEndpoint(t *testing.T) {
	cf := newStaticHTTPClientFactory([]staticHTTPResponse{
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]}`),
		},
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"members":[{"id":"2745e2525fce8fe","peerURLs":["http://127.0.0.1:7003"],"name":"node3","clientURLs":["http://127.0.0.1:4003"]},{"id":"42134f434382925","peerURLs":["http://127.0.0.1:2380","http://127.0.0.1:7001"],"name":"node1","clientURLs":["http://127.0.0.1:2379","http://127.0.0.1:4001"]},{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
		{
			resp: http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}},
			body: []byte(`{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"],"name":"node2","clientURLs":["http://127.0.0.1:4002"]}`),
		},
	})

	hc := &httpClusterClient{
		clientFactory: cf,
		rand:          rand.New(rand.NewSource(0)),
		selectionMode: EndpointSelectionPrioritizeLeader,
		endpoints:     []url.URL{{}}, // Need somewhere to pretend to send to initially
	}

	wants := []string{"http://127.0.0.1:4003", "http://127.0.0.1:4002"}

	for i, want := range wants {
		err := hc.Sync(context.Background())
		if err != nil {
			t.Fatalf("#%d: unexpected error during Sync: %#v", i, err)
		}

		pinned := hc.endpoints[hc.pinned].String()
		if pinned != want {
			t.Errorf("#%d: pinned endpoint = %v, want %v", i, pinned, want)
		}
	}
}

func TestHTTPClusterClientResetFail(t *testing.T) {
	tests := [][]string{
		// need at least one endpoint
		{},

		// urls must be valid
		{":"},
	}

	for i, tt := range tests {
		hc := &httpClusterClient{rand: rand.New(rand.NewSource(0))}
		err := hc.SetEndpoints(tt)
		if err == nil {
			t.Errorf("#%d: expected non-nil error", i)
		}
	}
}

func TestHTTPClusterClientResetPinRandom(t *testing.T) {
	round := 2000
	pinNum := 0
	for i := 0; i < round; i++ {
		hc := &httpClusterClient{rand: rand.New(rand.NewSource(int64(i)))}
		err := hc.SetEndpoints([]string{"http://127.0.0.1:4001", "http://127.0.0.1:4002", "http://127.0.0.1:4003"})
		if err != nil {
			t.Fatalf("#%d: reset error (%v)", i, err)
		}
		if hc.endpoints[hc.pinned].String() == "http://127.0.0.1:4001" {
			pinNum++
		}
	}

	min := 1.0/3.0 - 0.05
	max := 1.0/3.0 + 0.05
	if ratio := float64(pinNum) / float64(round); ratio > max || ratio < min {
		t.Errorf("pinned ratio = %v, want [%v, %v]", ratio, min, max)
	}
}