ext_test.go 6.97 KB
Newer Older
Jeromy's avatar
Jeromy committed
1 2 3
package dht

import (
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
4 5
	"io"
	"io/ioutil"
6
	"math/rand"
Jeromy's avatar
Jeromy committed
7
	"testing"
8
	"time"
Jeromy's avatar
Jeromy committed
9

10 11 12 13 14 15 16
	inet "github.com/ipfs/go-ipfs/p2p/net"
	mocknet "github.com/ipfs/go-ipfs/p2p/net/mock"
	peer "github.com/ipfs/go-ipfs/p2p/peer"
	routing "github.com/ipfs/go-ipfs/routing"
	pb "github.com/ipfs/go-ipfs/routing/dht/pb"
	record "github.com/ipfs/go-ipfs/routing/record"
	u "github.com/ipfs/go-ipfs/util"
Jeromy's avatar
Jeromy committed
17

18 19 20 21
	ggio "github.com/ipfs/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
	ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
	dssync "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
Jeromy's avatar
Jeromy committed
22 23
)

24
func TestGetFailures(t *testing.T) {
25 26 27
	if testing.Short() {
		t.SkipNow()
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
28

29
	ctx := context.Background()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
30
	mn, err := mocknet.FullMeshConnected(ctx, 2)
31 32 33
	if err != nil {
		t.Fatal(err)
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
34
	hosts := mn.Hosts()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
35

36
	tsds := dssync.MutexWrap(ds.NewMapDatastore())
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
37
	d := NewDHT(ctx, hosts[0], tsds)
38
	d.Update(ctx, hosts[1].ID())
39

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
40 41 42 43 44 45 46
	// u.POut("NotFound Test\n")
	// Reply with failures to every message
	hosts[1].SetStreamHandler(ProtocolDHT, func(s inet.Stream) {
		defer s.Close()
		io.Copy(ioutil.Discard, s)
	})

47
	// This one should time out
48
	// u.POut("Timout Test\n")
49
	ctx1, _ := context.WithTimeout(context.Background(), 200*time.Millisecond)
50
	if _, err := d.GetValue(ctx1, u.Key("test")); err != nil {
51 52 53 54
		if merr, ok := err.(u.MultiErr); ok && len(merr) > 0 {
			err = merr[0]
		}

55
		if err != context.DeadlineExceeded && err != context.Canceled {
56
			t.Fatal("Got different error than we expected", err)
57 58 59
		}
	} else {
		t.Fatal("Did not get expected error!")
Jeromy's avatar
Jeromy committed
60 61
	}

62
	t.Log("Timeout test passed.")
63

64
	// Reply with failures to every message
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
65
	hosts[1].SetStreamHandler(ProtocolDHT, func(s inet.Stream) {
66 67 68 69 70
		defer s.Close()

		pbr := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
		pbw := ggio.NewDelimitedWriter(s)

71
		pmes := new(pb.Message)
72 73
		if err := pbr.ReadMsg(pmes); err != nil {
			panic(err)
74 75
		}

76
		resp := &pb.Message{
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
77
			Type: pmes.Type,
78
		}
79 80 81
		if err := pbw.WriteMsg(resp); err != nil {
			panic(err)
		}
82 83
	})

84 85 86 87 88 89
	// This one should fail with NotFound.
	// long context timeout to ensure we dont end too early.
	// the dht should be exhausting its query and returning not found.
	// (was 3 seconds before which should be _plenty_ of time, but maybe
	// travis machines really have a hard time...)
	ctx2, _ := context.WithTimeout(context.Background(), 20*time.Second)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
90
	_, err = d.GetValue(ctx2, u.Key("test"))
91
	if err != nil {
92 93 94
		if merr, ok := err.(u.MultiErr); ok && len(merr) > 0 {
			err = merr[0]
		}
95
		if err != routing.ErrNotFound {
96
			t.Fatalf("Expected ErrNotFound, got: %s", err)
97 98 99 100
		}
	} else {
		t.Fatal("expected error, got none.")
	}
101

102 103
	t.Log("ErrNotFound check passed!")

104
	// Now we test this DHT's handleGetValue failure
105 106 107
	{
		typ := pb.Message_GET_VALUE
		str := "hello"
Jeromy's avatar
Jeromy committed
108 109 110 111 112 113

		sk, err := d.getOwnPrivateKey()
		if err != nil {
			t.Fatal(err)
		}

Jeromy's avatar
Jeromy committed
114
		rec, err := record.MakePutRecord(sk, u.Key(str), []byte("blah"), true)
115 116 117 118 119 120 121 122
		if err != nil {
			t.Fatal(err)
		}
		req := pb.Message{
			Type:   &typ,
			Key:    &str,
			Record: rec,
		}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
123

124
		// u.POut("handleGetValue Test\n")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
125
		s, err := hosts[1].NewStream(ProtocolDHT, hosts[0].ID())
126 127 128 129
		if err != nil {
			t.Fatal(err)
		}
		defer s.Close()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
130

131 132
		pbr := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
		pbw := ggio.NewDelimitedWriter(s)
133

134 135 136
		if err := pbw.WriteMsg(&req); err != nil {
			t.Fatal(err)
		}
137

138 139 140 141 142 143 144 145 146 147 148
		pmes := new(pb.Message)
		if err := pbr.ReadMsg(pmes); err != nil {
			t.Fatal(err)
		}
		if pmes.GetRecord() != nil {
			t.Fatal("shouldnt have value")
		}
		if pmes.GetProviderPeers() != nil {
			t.Fatal("shouldnt have provider peers")
		}
	}
Jeromy's avatar
Jeromy committed
149
}
Jeromy's avatar
Jeromy committed
150 151

func TestNotFound(t *testing.T) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
152
	// t.Skip("skipping test to debug another")
153 154 155
	if testing.Short() {
		t.SkipNow()
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
156

157
	ctx := context.Background()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
158
	mn, err := mocknet.FullMeshConnected(ctx, 16)
159 160 161
	if err != nil {
		t.Fatal(err)
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
162
	hosts := mn.Hosts()
163
	tsds := dssync.MutexWrap(ds.NewMapDatastore())
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
164
	d := NewDHT(ctx, hosts[0], tsds)
Jeromy's avatar
Jeromy committed
165

166 167
	for _, p := range hosts {
		d.Update(ctx, p.ID())
Jeromy's avatar
Jeromy committed
168 169 170
	}

	// Reply with random peers to every message
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
171 172 173
	for _, host := range hosts {
		host := host // shadow loop var
		host.SetStreamHandler(ProtocolDHT, func(s inet.Stream) {
174
			defer s.Close()
Jeromy's avatar
Jeromy committed
175

176 177
			pbr := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
			pbw := ggio.NewDelimitedWriter(s)
Jeromy's avatar
Jeromy committed
178

179 180 181
			pmes := new(pb.Message)
			if err := pbr.ReadMsg(pmes); err != nil {
				panic(err)
Jeromy's avatar
Jeromy committed
182 183
			}

184 185 186 187
			switch pmes.GetType() {
			case pb.Message_GET_VALUE:
				resp := &pb.Message{Type: pmes.Type}

188
				ps := []peer.PeerInfo{}
189
				for i := 0; i < 7; i++ {
190
					p := hosts[rand.Intn(len(hosts))].ID()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
191
					pi := host.Peerstore().PeerInfo(p)
192
					ps = append(ps, pi)
193 194
				}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
195
				resp.CloserPeers = pb.PeerInfosToPBPeers(d.host.Network(), ps)
196 197 198 199 200 201 202 203 204
				if err := pbw.WriteMsg(resp); err != nil {
					panic(err)
				}

			default:
				panic("Shouldnt recieve this.")
			}
		})
	}
Jeromy's avatar
Jeromy committed
205

206 207
	// long timeout to ensure timing is not at play.
	ctx, _ = context.WithTimeout(ctx, time.Second*20)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
208
	v, err := d.GetValue(ctx, u.Key("hello"))
209
	log.Debugf("get value got %v", v)
Jeromy's avatar
Jeromy committed
210
	if err != nil {
211 212 213
		if merr, ok := err.(u.MultiErr); ok && len(merr) > 0 {
			err = merr[0]
		}
Jeromy's avatar
Jeromy committed
214
		switch err {
215
		case routing.ErrNotFound:
Jeromy's avatar
Jeromy committed
216 217 218 219 220 221 222 223 224 225
			//Success!
			return
		case u.ErrTimeout:
			t.Fatal("Should not have gotten timeout!")
		default:
			t.Fatalf("Got unexpected error: %s", err)
		}
	}
	t.Fatal("Expected to recieve an error.")
}
226 227 228 229

// If less than K nodes are in the entire network, it should fail when we make
// a GET rpc and nobody has the value
func TestLessThanKResponses(t *testing.T) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
230
	// t.Skip("skipping test to debug another")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
231 232
	// t.Skip("skipping test because it makes a lot of output")

233
	ctx := context.Background()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
234
	mn, err := mocknet.FullMeshConnected(ctx, 6)
235 236 237
	if err != nil {
		t.Fatal(err)
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
238
	hosts := mn.Hosts()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
239

240
	tsds := dssync.MutexWrap(ds.NewMapDatastore())
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
241
	d := NewDHT(ctx, hosts[0], tsds)
242

243
	for i := 1; i < 5; i++ {
244
		d.Update(ctx, hosts[i].ID())
245 246 247
	}

	// Reply with random peers to every message
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
248 249 250
	for _, host := range hosts {
		host := host // shadow loop var
		host.SetStreamHandler(ProtocolDHT, func(s inet.Stream) {
251 252 253 254
			defer s.Close()

			pbr := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
			pbw := ggio.NewDelimitedWriter(s)
255

256 257 258
			pmes := new(pb.Message)
			if err := pbr.ReadMsg(pmes); err != nil {
				panic(err)
259 260
			}

261 262
			switch pmes.GetType() {
			case pb.Message_GET_VALUE:
263
				pi := host.Peerstore().PeerInfo(hosts[1].ID())
264 265
				resp := &pb.Message{
					Type:        pmes.Type,
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
266
					CloserPeers: pb.PeerInfosToPBPeers(d.host.Network(), []peer.PeerInfo{pi}),
267 268 269 270 271 272 273
				}

				if err := pbw.WriteMsg(resp); err != nil {
					panic(err)
				}
			default:
				panic("Shouldnt recieve this.")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
274
			}
275

276 277
		})
	}
278

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
279
	ctx, _ = context.WithTimeout(ctx, time.Second*30)
280
	if _, err := d.GetValue(ctx, u.Key("hello")); err != nil {
281
		switch err {
282
		case routing.ErrNotFound:
283 284 285 286 287 288 289 290 291 292
			//Success!
			return
		case u.ErrTimeout:
			t.Fatal("Should not have gotten timeout!")
		default:
			t.Fatalf("Got unexpected error: %s", err)
		}
	}
	t.Fatal("Expected to recieve an error.")
}