func (d *discovery) checkCluster() ([]*client.Node, int, uint64, error) {
	configKey := path.Join("/", d.cluster, "_config")
	ctx, cancel := context.WithTimeout(context.Background(), client.DefaultRequestTimeout)
	// find cluster size
	resp, err := d.c.Get(ctx, path.Join(configKey, "size"), nil)
	cancel()
	if err != nil {
		if eerr, ok := err.(*client.Error); ok && eerr.Code == client.ErrorCodeKeyNotFound {
			return nil, 0, 0, ErrSizeNotFound
		}
		if err == context.DeadlineExceeded {
			return d.checkClusterRetry()
		}
		return nil, 0, 0, err
	}
	size, err := strconv.Atoi(resp.Node.Value)
	if err != nil {
		return nil, 0, 0, ErrBadSizeKey
	}

	ctx, cancel = context.WithTimeout(context.Background(), client.DefaultRequestTimeout)
	resp, err = d.c.Get(ctx, d.cluster, nil)
	cancel()
	if err != nil {
		if err == context.DeadlineExceeded {
			return d.checkClusterRetry()
		}
		return nil, 0, 0, err
	}
	nodes := make([]*client.Node, 0)
	// append non-config keys to nodes
	for _, n := range resp.Node.Nodes {
		if !(path.Base(n.Key) == path.Base(configKey)) {
			nodes = append(nodes, n)
		}
	}

	snodes := sortableNodes{nodes}
	sort.Sort(snodes)

	// find self position
	for i := range nodes {
		if path.Base(nodes[i].Key) == path.Base(d.selfKey()) {
			break
		}
		if i >= size-1 {
			return nodes[:size], size, resp.Index, ErrFullCluster
		}
	}
	return nodes, size, resp.Index, nil
}

// publish registers server information into the cluster. The information
// is the JSON representation of this server's member struct, updated with the
// static clientURLs of the server.
// The function keeps attempting to register until it succeeds,
// or its server is stopped.
func (s *EtcdServer) publish(retryInterval time.Duration) {
	b, err := json.Marshal(s.attributes)
	if err != nil {
		log.Printf("etcdserver: json marshal error: %v", err)
		return
	}
	req := pb.Request{
		Method: "PUT",
		Path:   MemberAttributesStorePath(s.id),
		Val:    string(b),
	}

	for {
		ctx, cancel := context.WithTimeout(context.Background(), retryInterval)
		_, err := s.Do(ctx, req)
		cancel()
		switch err {
		case nil:
			log.Printf("etcdserver: published %+v to cluster %s", s.attributes, s.Cluster.ID())
			return
		case ErrStopped:
			log.Printf("etcdserver: aborting publish because server is stopped")
			return
		default:
			log.Printf("etcdserver: publish error: %v", err)
		}
	}
}

func TestHTTPKeysAPISetResponse(t *testing.T) {
	client := &staticHTTPClient{
		resp: http.Response{
			StatusCode: http.StatusOK,
			Header:     http.Header{"X-Etcd-Index": []string{"21"}},
		},
		body: []byte(`{"action":"set","node":{"key":"/pants/foo/bar/baz","value":"snarf","modifiedIndex":21,"createdIndex":21},"prevNode":{"key":"/pants/foo/bar/baz","value":"snazz","modifiedIndex":20,"createdIndex":19}}`),
	}

	wantResponse := &Response{
		Action:   "set",
		Node:     &Node{Key: "/pants/foo/bar/baz", Value: "snarf", CreatedIndex: uint64(21), ModifiedIndex: uint64(21)},
		PrevNode: &Node{Key: "/pants/foo/bar/baz", Value: "snazz", CreatedIndex: uint64(19), ModifiedIndex: uint64(20)},
		Index:    uint64(21),
	}

	kAPI := &httpKeysAPI{client: client, prefix: "/pants"}
	resp, err := kAPI.Set(context.Background(), "/foo/bar/baz", "snarf", nil)
	if err != nil {
		t.Errorf("non-nil error: %#v", err)
	}
	if !reflect.DeepEqual(wantResponse, resp) {
		t.Errorf("incorrect Response: want=%#v got=%#v", wantResponse, resp)
	}
}

func TestHTTPMembersAPIAddSuccess(t *testing.T) {
	wantAction := &membersAPIActionAdd{
		peerURLs: types.URLs([]url.URL{
			url.URL{Scheme: "http", Host: "127.0.0.1:7002"},
		}),
	}

	mAPI := &httpMembersAPI{
		client: &actionAssertingHTTPClient{
			t:   t,
			act: wantAction,
			resp: http.Response{
				StatusCode: http.StatusCreated,
			},
			body: []byte(`{"id":"94088180e21eb87b","peerURLs":["http://127.0.0.1:7002"]}`),
		},
	}

	wantResponseMember := &Member{
		ID:       "94088180e21eb87b",
		PeerURLs: []string{"http://127.0.0.1:7002"},
	}

	m, err := mAPI.Add(context.Background(), "http://127.0.0.1:7002")
	if err != nil {
		t.Errorf("got non-nil err: %#v", err)
	}
	if !reflect.DeepEqual(wantResponseMember, m) {
		t.Errorf("incorrect Member: want=%#v got=%#v", wantResponseMember, m)
	}
}

func TestHTTPMembersAPIListSuccess(t *testing.T) {
	wantAction := &membersAPIActionList{}
	mAPI := &httpMembersAPI{
		client: &actionAssertingHTTPClient{
			t:   t,
			act: wantAction,
			resp: http.Response{
				StatusCode: http.StatusOK,
			},
			body: []byte(`{"members":[{"id":"94088180e21eb87b","name":"node2","peerURLs":["http://127.0.0.1:7002"],"clientURLs":["http://127.0.0.1:4002"]}]}`),
		},
	}

	wantResponseMembers := []Member{
		Member{
			ID:         "94088180e21eb87b",
			Name:       "node2",
			PeerURLs:   []string{"http://127.0.0.1:7002"},
			ClientURLs: []string{"http://127.0.0.1:4002"},
		},
	}

	m, err := mAPI.List(context.Background())
	if err != nil {
		t.Errorf("got non-nil err: %#v", err)
	}
	if !reflect.DeepEqual(wantResponseMembers, m) {
		t.Errorf("incorrect Members: want=%#v got=%#v", wantResponseMembers, m)
	}
}

func TestHTTPMembersAPIListError(t *testing.T) {
	tests := []httpClient{
		// generic httpClient failure
		&staticHTTPClient{err: errors.New("fail!")},

		// unrecognized HTTP status code
		&staticHTTPClient{
			resp: http.Response{StatusCode: http.StatusTeapot},
		},

		// fail to unmarshal body on StatusOK
		&staticHTTPClient{
			resp: http.Response{
				StatusCode: http.StatusOK,
			},
			body: []byte(`[{"id":"XX`),
		},
	}

	for i, tt := range tests {
		mAPI := &httpMembersAPI{client: tt}
		ms, err := mAPI.List(context.Background())
		if err == nil {
			t.Errorf("#%d: got nil err", i)
		}
		if ms != nil {
			t.Errorf("#%d: got non-nil Member slice", i)
		}
	}
}

func TestHTTPKeysAPIGetResponse(t *testing.T) {
	client := &staticHTTPClient{
		resp: http.Response{
			StatusCode: http.StatusOK,
			Header:     http.Header{"X-Etcd-Index": []string{"42"}},
		},
		body: []byte(`{"action":"get","node":{"key":"/pants/foo/bar","modifiedIndex":25,"createdIndex":19,"nodes":[{"key":"/pants/foo/bar/baz","value":"snarf","createdIndex":21,"modifiedIndex":25}]}}`),
	}

	wantResponse := &Response{
		Action: "get",
		Node: &Node{
			Key: "/pants/foo/bar",
			Nodes: []*Node{
				&Node{Key: "/pants/foo/bar/baz", Value: "snarf", CreatedIndex: 21, ModifiedIndex: 25},
			},
			CreatedIndex:  uint64(19),
			ModifiedIndex: uint64(25),
		},
		Index: uint64(42),
	}

	kAPI := &httpKeysAPI{client: client, prefix: "/pants"}
	resp, err := kAPI.Get(context.Background(), "/foo/bar", &GetOptions{Recursive: true})
	if err != nil {
		t.Errorf("non-nil error: %#v", err)
	}
	if !reflect.DeepEqual(wantResponse, resp) {
		t.Errorf("incorrect Response: want=%#v got=%#v", wantResponse, resp)
	}
}

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

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

	want := []string{"http://127.0.0.1:4001"}
	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 (d *discovery) createSelf(contents string) error {
	ctx, cancel := context.WithTimeout(context.Background(), client.DefaultRequestTimeout)
	resp, err := d.c.Create(ctx, d.selfKey(), contents)
	cancel()
	if err != nil {
		if eerr, ok := err.(*client.Error); ok && eerr.Code == client.ErrorCodeNodeExist {
			return ErrDuplicateID
		}
		return err
	}

	// ensure self appears on the server we connected to
	w := d.c.Watcher(d.selfKey(), &client.WatcherOptions{AfterIndex: resp.Node.CreatedIndex - 1})
	_, err = w.Next(context.Background())
	return err
}

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 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 for CancelRequest to be called, informing us that simpleHTTPClient
		// knows the context is already timed out
		<-tr.startCancel

		tr.respchan <- &http.Response{Body: body}
		tr.finishCancel <- struct{}{}
	}()

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

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

func (d *discovery) waitNodes(nodes []*client.Node, size int, index uint64) ([]*client.Node, error) {
	if len(nodes) > size {
		nodes = nodes[:size]
	}
	// watch from the next index
	w := d.c.Watcher(d.cluster, &client.WatcherOptions{AfterIndex: index, Recursive: true})
	all := make([]*client.Node, len(nodes))
	copy(all, nodes)
	for _, n := range all {
		if path.Base(n.Key) == path.Base(d.selfKey()) {
			log.Printf("discovery: found self %s in the cluster", path.Base(d.selfKey()))
		} else {
			log.Printf("discovery: found peer %s in the cluster", path.Base(n.Key))
		}
	}

	// wait for others
	for len(all) < size {
		log.Printf("discovery: found %d peer(s), waiting for %d more", len(all), size-len(all))
		resp, err := w.Next(context.Background())
		if err != nil {
			if err == context.DeadlineExceeded {
				return d.waitNodesRetry()
			}
			return nil, err
		}
		log.Printf("discovery: found peer %s in the cluster", path.Base(resp.Node.Key))
		all = append(all, resp.Node)
	}
	log.Printf("discovery: found %d needed peer(s)", len(all))
	return all, nil
}

func TestHTTPKeysAPIGetError(t *testing.T) {
	tests := []httpClient{
		// generic HTTP client failure
		&staticHTTPClient{
			err: errors.New("fail!"),
		},

		// unusable status code
		&staticHTTPClient{
			resp: http.Response{
				StatusCode: http.StatusTeapot,
			},
		},

		// etcd Error response
		&staticHTTPClient{
			resp: http.Response{
				StatusCode: http.StatusInternalServerError,
			},
			body: []byte(`{"errorCode":300,"message":"Raft internal error","cause":"/foo","index":18}`),
		},
	}

	for i, tt := range tests {
		kAPI := httpKeysAPI{client: tt}
		resp, err := kAPI.Get(context.Background(), "/foo", nil)
		if err == nil {
			t.Errorf("#%d: received nil error", i)
		}
		if resp != nil {
			t.Errorf("#%d: received non-nil Response: %#v", i, resp)
		}
	}
}

// TestMultiNodeStart ensures that a node can be started correctly. The node should
// start with correct configuration change entries, and can accept and commit
// proposals.
func TestMultiNodeStart(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
	ccdata, err := cc.Marshal()
	if err != nil {
		t.Fatalf("unexpected marshal error: %v", err)
	}
	wants := []Ready{
		{
			SoftState: &SoftState{Lead: 1, RaftState: StateLeader},
			HardState: raftpb.HardState{Term: 2, Commit: 2, Vote: 1},
			Entries: []raftpb.Entry{
				{Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
				{Term: 2, Index: 2},
			},
			CommittedEntries: []raftpb.Entry{
				{Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
				{Term: 2, Index: 2},
			},
		},
		{
			HardState:        raftpb.HardState{Term: 2, Commit: 3, Vote: 1},
			Entries:          []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
			CommittedEntries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
		},
	}
	mn := StartMultiNode(1, 10, 1)
	storage := NewMemoryStorage()
	mn.CreateGroup(1, []Peer{{ID: 1}}, storage)
	mn.Campaign(ctx, 1)
	gs := <-mn.Ready()
	g := gs[1]
	if !reflect.DeepEqual(g, wants[0]) {
		t.Fatalf("#%d: g = %+v,\n             w   %+v", 1, g, wants[0])
	} else {
		storage.Append(g.Entries)
		mn.Advance(gs)
	}

	mn.Propose(ctx, 1, []byte("foo"))
	if gs2 := <-mn.Ready(); !reflect.DeepEqual(gs2[1], wants[1]) {
		t.Errorf("#%d: g = %+v,\n             w   %+v", 2, gs2[1], wants[1])
	} else {
		storage.Append(gs2[1].Entries)
		mn.Advance(gs2)
	}

	select {
	case rd := <-mn.Ready():
		t.Errorf("unexpected Ready: %+v", rd)
	case <-time.After(time.Millisecond):
	}
}

func (h *keysHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if !allowMethod(w, r.Method, "HEAD", "GET", "PUT", "POST", "DELETE") {
		return
	}

	w.Header().Set("X-Etcd-Cluster-ID", h.clusterInfo.ID().String())

	ctx, cancel := context.WithTimeout(context.Background(), h.timeout)
	defer cancel()

	rr, err := parseKeyRequest(r, clockwork.NewRealClock())
	if err != nil {
		writeError(w, err)
		return
	}
	// The path must be valid at this point (we've parsed the request successfully).
	if !hasKeyPrefixAccess(h.sec, r, r.URL.Path[len(keysPrefix):]) {
		writeNoAuth(w)
		return
	}

	resp, err := h.server.Do(ctx, rr)
	if err != nil {
		err = trimErrorPrefix(err, etcdserver.StoreKeysPrefix)
		writeError(w, err)
		return
	}

	switch {
	case resp.Event != nil:
		if err := writeKeyEvent(w, resp.Event, h.timer); err != nil {
			// Should never be reached
			log.Printf("error writing event: %v", err)
		}
	case resp.Watcher != nil:
		ctx, cancel := context.WithTimeout(context.Background(), defaultWatchTimeout)
		defer cancel()
		handleKeyWatch(ctx, w, resp.Watcher, rr.Stream, h.timer)
	default:
		writeError(w, errors.New("received response with no Event/Watcher!"))
	}
}

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 (s *Store) requestResource(res string, dir bool) (etcdserver.Response, error) {
	ctx, cancel := context.WithTimeout(context.Background(), s.timeout)
	defer cancel()
	p := path.Join(StorePermsPrefix, res)
	rr := etcdserverpb.Request{
		Method: "GET",
		Path:   p,
		Dir:    dir,
	}
	return s.server.Do(ctx, rr)
}

func (s *Store) deleteResource(res string) (etcdserver.Response, error) {
	ctx, cancel := context.WithTimeout(context.Background(), s.timeout)
	defer cancel()
	pex := true
	p := path.Join(StorePermsPrefix, res)
	rr := etcdserverpb.Request{
		Method:    "DELETE",
		Path:      p,
		PrevExist: &pex,
	}
	return s.server.Do(ctx, rr)
}

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")
	}
}

func TestHTTPKeysAPIDeleteAction(t *testing.T) {
	tests := []struct {
		key        string
		value      string
		opts       *DeleteOptions
		wantAction httpAction
	}{
		// nil DeleteOptions
		{
			key:  "/foo",
			opts: nil,
			wantAction: &deleteAction{
				Key:       "/foo",
				PrevValue: "",
				PrevIndex: 0,
				Recursive: false,
			},
		},
		// empty DeleteOptions
		{
			key:  "/foo",
			opts: &DeleteOptions{},
			wantAction: &deleteAction{
				Key:       "/foo",
				PrevValue: "",
				PrevIndex: 0,
				Recursive: false,
			},
		},
		// populated DeleteOptions
		{
			key: "/foo",
			opts: &DeleteOptions{
				PrevValue: "baz",
				PrevIndex: 13,
				Recursive: true,
			},
			wantAction: &deleteAction{
				Key:       "/foo",
				PrevValue: "baz",
				PrevIndex: 13,
				Recursive: true,
			},
		},
	}

	for i, tt := range tests {
		client := &actionAssertingHTTPClient{t: t, num: i, act: tt.wantAction}
		kAPI := httpKeysAPI{client: client}
		kAPI.Delete(context.Background(), tt.key, tt.opts)
	}
}