swarm/network: Added a bunch of (failing) streamer unit tests

swarm/network/streamer_test.go

@@ -17,12 +17,14 @@
 package network
 
 import (
+	"bytes"
 	"errors"
 	"io/ioutil"
 	"os"
 	"testing"
 	"time"
 
+	sha3 "github.com/ethereum/go-ethereum/crypto/sha3"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
@@ -35,7 +37,7 @@ func init() {
 }
 
 // TODO: extract newStreamer
-func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, func(), error) {
+func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, *storage.LocalStore, func(), error) {
 	// setup
 	addr := RandomAddr() // tested peers peer address
 	to := NewKademlia(addr.OAddr, NewKadParams())
@@ -43,7 +45,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, func()
 	// temp datadir
 	datadir, err := ioutil.TempDir("", "streamer")
 	if err != nil {
-		return nil, nil, func() {}, err
+		return nil, nil, nil, func() {}, err
 	}
 	teardown := func() {
 		os.RemoveAll(datadir)
@@ -51,7 +53,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, func()
 
 	localStore, err := storage.NewTestLocalStore(datadir)
 	if err != nil {
-		return nil, nil, teardown, err
+		return nil, nil, nil, teardown, err
 	}
 
 	dbAccess := NewDbAccess(localStore)
@@ -63,11 +65,12 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, func()
 			localAddr: addr,
 			BzzAddr:   NewAddrFromNodeID(p.ID()),
 		}
+		to.On(bzzPeer)
 		return streamer.Run(bzzPeer)
 	}
 
 	protocolTester := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), 1, run)
-	return protocolTester, streamer, teardown, nil
+	return protocolTester, streamer, localStore, teardown, nil
 }
 
 // TODO
@@ -76,7 +79,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, func()
 // }
 
 func TestStreamerSubscribe(t *testing.T) {
-	tester, streamer, teardown, err := newStreamerTester(t)
+	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()
 	if err != nil {
 		t.Fatal(err)
@@ -88,6 +91,18 @@ func TestStreamerSubscribe(t *testing.T) {
 	}
 }
 
+var (
+	hash0                            = sha3.Sum256([]byte{0})
+	hash1                            = sha3.Sum256([]byte{1})
+	hash2                            = sha3.Sum256([]byte{2})
+	hashesTmp                        = append(hash0[:], hash1[:]...)
+	hashes                           = append(hashesTmp, hash2[:]...)
+	receivedHashes map[string][]byte = make(map[string][]byte)
+	wait0                            = make(chan bool)
+	wait2                            = make(chan bool)
+	batchDone                        = make(chan bool)
+)
+
 type testIncomingStreamer struct {
 	t []byte
 }
@@ -96,11 +111,22 @@ type testOutgoingStreamer struct {
 	t []byte
 }
 
-func (self *testIncomingStreamer) NeedData([]byte) func() {
+func (self *testIncomingStreamer) NeedData(hash []byte) func() {
+	receivedHashes[string(hash)] = hash
+	if bytes.Equal(hash, hash0[:]) {
+		return func() {
+			<-wait0
+		}
+	} else if bytes.Equal(hash, hash2[:]) {
+		return func() {
+			<-wait2
+		}
+	}
 	return nil
 }
 
 func (self *testIncomingStreamer) BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error) {
+	close(batchDone)
 	return nil
 }
 
@@ -112,9 +138,9 @@ func (self *testOutgoingStreamer) GetData([]byte) []byte {
 	return nil
 }
 
-func TestStreamerRegisterIncoming(t *testing.T) {
+func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) {
 	// TODO: we only need streamer
-	tester, streamer, teardown, err := newStreamerTester(t)
+	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()
 	if err != nil {
 		t.Fatal(err)
@@ -160,9 +186,9 @@ func TestStreamerRegisterIncoming(t *testing.T) {
 	}
 }
 
-func TestStreamerRegisterOutgoing(t *testing.T) {
+func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) {
 	// TODO: we only need streamer
-	tester, streamer, teardown, err := newStreamerTester(t)
+	tester, streamer, _, teardown, err := newStreamerTester(t)
 	defer teardown()
 	if err != nil {
 		t.Fatal(err)
@@ -210,6 +236,323 @@ func TestStreamerRegisterOutgoing(t *testing.T) {
 		},
 	})
 
+	if err != nil {
+		t.Fatal(err)
+	}
+
+}
+
+func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
+	// TODO: we only need streamer
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	streamer.RegisterIncomingStreamer("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+		return &testIncomingStreamer{
+			t: t,
+		}, nil
+	})
+
+	err = waitForPeers(streamer, 1*time.Second)
+	if err != nil {
+		t.Fatal("timeout: peer is not created")
+	}
+
+	peerId := tester.IDs[0]
+
+	err = streamer.Subscribe(peerId, "foo", []byte{}, 5, 8, Top, true)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "Subscribe message",
+		Expects: []p2ptest.Expect{
+			p2ptest.Expect{
+				Code: 4,
+				Msg: &SubscribeMsg{
+					Stream:   "foo",
+					Key:      []byte{},
+					From:     5,
+					To:       8,
+					Priority: Top,
+				},
+				Peer: peerId,
+			},
+		},
+	},
+		p2ptest.Exchange{
+			Label: "WantedHashes message",
+			Triggers: []p2ptest.Trigger{
+				p2ptest.Trigger{
+					Code: 1,
+					Msg: &OfferedHashesMsg{
+						HandoverProof: &HandoverProof{
+							Handover: &Handover{},
+						},
+						Hashes: hashes,
+						From:   5,
+						To:     8,
+						Stream: "foo",
+					},
+					Peer: peerId,
+				},
+			},
+			Expects: []p2ptest.Expect{
+				p2ptest.Expect{
+					Code: 2,
+					Msg: &WantedHashesMsg{
+						Stream: "foo",
+						Want:   []byte{5},
+						From:   8,
+						To:     0,
+					},
+					Peer: peerId,
+				},
+			},
+		})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(receivedHashes) != 3 {
+		t.Fatalf("Expected number of received hashes %v, got %v", 3, len(receivedHashes))
+	}
+
+	close(wait0)
+
+	timeout := time.NewTimer(100 * time.Millisecond)
+	defer timeout.Stop()
+
+	select {
+	case <-batchDone:
+		t.Fatal("batch done early")
+	case <-timeout.C:
+	}
+
+	close(wait2)
+
+	timeout2 := time.NewTimer(10000 * time.Millisecond)
+	defer timeout2.Stop()
+
+	select {
+	case <-batchDone:
+	case <-timeout2.C:
+		t.Fatal("timeout waiting batchdone call")
+	}
+
+}
+
+func TestRetrieveRequest(t *testing.T) {
+	// TODO: we only need streamer
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = waitForPeers(streamer, 1*time.Second)
+	if err != nil {
+		t.Fatal("timeout: peer is not created")
+	}
+
+	peerId := tester.IDs[0]
+
+	chunk := storage.NewChunk(storage.Key(hash0[:]), nil)
+
+	streamer.Retrieve(chunk)
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "RetrieveRequestMsg",
+		Expects: []p2ptest.Expect{
+			p2ptest.Expect{
+				Code: 5,
+				Msg: &RetrieveRequestMsg{
+					Key: chunk.Key[:],
+				},
+				Peer: peerId,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+}
+
+func TestUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) {
+	// TODO: we only need streamer
+	tester, streamer, _, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// streamer.RegisterOutgoingStreamer("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+	// 	return &testOutgoingStreamer{
+	// 		t: t,
+	// 	}, nil
+	// })
+
+	err = waitForPeers(streamer, 1*time.Second)
+	if err != nil {
+		t.Fatal("timeout: peer is not created")
+	}
+
+	peerId := tester.IDs[0]
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "SubscribeMsg",
+		Expects: []p2ptest.Expect{
+			p2ptest.Expect{
+				Code: 4,
+				Msg: &SubscribeMsg{
+					Stream:   retrieveRequestStream,
+					Key:      nil,
+					From:     0,
+					To:       0,
+					Priority: Top,
+				},
+				Peer: peerId,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	chunk := storage.NewChunk(storage.Key(hash0[:]), nil)
+
+	peer := streamer.getPeer(peerId)
+
+	peer.handleSubscribeMsg(&SubscribeMsg{
+		Stream:   retrieveRequestStream,
+		Key:      nil,
+		From:     0,
+		To:       0,
+		Priority: Top,
+	})
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "RetrieveRequestMsg",
+		Triggers: []p2ptest.Trigger{
+			p2ptest.Trigger{
+				Code: 5,
+				Msg: &RetrieveRequestMsg{
+					Key: chunk.Key[:],
+				},
+				Peer: peerId,
+			},
+		},
+		Expects: []p2ptest.Expect{
+			p2ptest.Expect{
+				Code: 1,
+				Msg: &OfferedHashesMsg{
+					HandoverProof: nil,
+					Hashes:        nil,
+					From:          0,
+					To:            0,
+				},
+				Peer: peerId,
+			},
+		},
+	})
+
+	expectedError := "exchange 0: 'RetrieveRequestMsg' timed out"
+	if err == nil || err.Error() != expectedError {
+		t.Fatalf("Expected error %v, got %v", expectedError, err)
+	}
+}
+
+func TestUpstreamRetrieveRequestMsgExchange(t *testing.T) {
+	// TODO: we only need streamer
+	tester, streamer, localStore, teardown, err := newStreamerTester(t)
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// streamer.RegisterOutgoingStreamer("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+	// 	return &testOutgoingStreamer{
+	// 		t: t,
+	// 	}, nil
+	// })
+
+	err = waitForPeers(streamer, 1*time.Second)
+	if err != nil {
+		t.Fatal("timeout: peer is not created")
+	}
+
+	peerId := tester.IDs[0]
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "SubscribeMsg",
+		Expects: []p2ptest.Expect{
+			p2ptest.Expect{
+				Code: 4,
+				Msg: &SubscribeMsg{
+					Stream:   retrieveRequestStream,
+					Key:      nil,
+					From:     0,
+					To:       0,
+					Priority: Top,
+				},
+				Peer: peerId,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	chunk := storage.NewChunk(storage.Key(hash0[:]), nil)
+
+	peer := streamer.getPeer(peerId)
+
+	peer.handleSubscribeMsg(&SubscribeMsg{
+		Stream:   retrieveRequestStream,
+		Key:      nil,
+		From:     0,
+		To:       0,
+		Priority: Top,
+	})
+
+	chunk.SData = hash0[:]
+	localStore.Put(chunk)
+
+	err = tester.TestExchanges(p2ptest.Exchange{
+		Label: "RetrieveRequestMsg",
+		Triggers: []p2ptest.Trigger{
+			p2ptest.Trigger{
+				Code: 5,
+				Msg: &RetrieveRequestMsg{
+					Key: chunk.Key[:],
+				},
+				Peer: peerId,
+			},
+		},
+		Expects: []p2ptest.Expect{
+			p2ptest.Expect{
+				Code: 1,
+				Msg: &OfferedHashesMsg{
+					HandoverProof: nil,
+					Hashes:        chunk.Key[:],
+					From:          0,
+					// TODO: why is this 32???
+					To:     32,
+					Key:    []byte{},
+					Stream: retrieveRequestStream,
+				},
+				Peer: peerId,
+			},
+		},
+	})
+
 	if err != nil {
 		t.Fatal(err)
 	}