refactor(mdag, bserv, bs) mocks, etc.

License: MIT
Signed-off-by: Brian Tiger Chow <brian@perfmode.com>

blockservice/blocks_test.go

@@ -11,10 +11,7 @@ import (
 	blocks "github.com/jbenet/go-ipfs/blocks"
 	blockstore "github.com/jbenet/go-ipfs/blocks/blockstore"
 	blocksutil "github.com/jbenet/go-ipfs/blocks/blocksutil"
-	bitswap "github.com/jbenet/go-ipfs/exchange/bitswap"
-	tn "github.com/jbenet/go-ipfs/exchange/bitswap/testnet"
 	offline "github.com/jbenet/go-ipfs/exchange/offline"
-	"github.com/jbenet/go-ipfs/routing/mock"
 	u "github.com/jbenet/go-ipfs/util"
 )
 
@@ -63,23 +60,9 @@ func TestBlocks(t *testing.T) {
 }
 
 func TestGetBlocksSequential(t *testing.T) {
-	net := tn.VirtualNetwork()
+	var servs = Mocks(t, 4)
-	rs := mock.VirtualRoutingServer()
-	sg := bitswap.NewSessionGenerator(net, rs)
 	bg := blocksutil.NewBlockGenerator()
-
-	instances := sg.Instances(4)
 	blks := bg.Blocks(50)
-	// TODO: verify no duplicates
-
-	var servs []*BlockService
-	for _, i := range instances {
-		bserv, err := New(i.Blockstore, i.Exchange)
-		if err != nil {
-			t.Fatal(err)
-		}
-		servs = append(servs, bserv)
-	}
 
 	var keys []u.Key
 	for _, blk := range blks {
@@ -89,7 +72,7 @@ func TestGetBlocksSequential(t *testing.T) {
 
 	t.Log("one instance at a time, get blocks concurrently")
 
-	for i := 1; i < len(instances); i++ {
+	for i := 1; i < len(servs); i++ {
 		ctx, _ := context.WithTimeout(context.TODO(), time.Second*5)
 		out := servs[i].GetBlocks(ctx, keys)
 		gotten := make(map[u.Key]*blocks.Block)

blockservice/mock.go

@@ -0,0 +1,28 @@
+package blockservice
+
+import (
+	"testing"
+
+	bitswap "github.com/jbenet/go-ipfs/exchange/bitswap"
+	tn "github.com/jbenet/go-ipfs/exchange/bitswap/testnet"
+	"github.com/jbenet/go-ipfs/routing/mock"
+)
+
+// Mocks returns |n| connected mock Blockservices
+func Mocks(t *testing.T, n int) []*BlockService {
+	net := tn.VirtualNetwork()
+	rs := mock.VirtualRoutingServer()
+	sg := bitswap.NewSessionGenerator(net, rs)
+
+	instances := sg.Instances(n)
+
+	var servs []*BlockService
+	for _, i := range instances {
+		bserv, err := New(i.Blockstore(), i.Exchange)
+		if err != nil {
+			t.Fatal(err)
+		}
+		servs = append(servs, bserv)
+	}
+	return servs
+}

core/core.go

@@ -28,10 +28,10 @@ import (
 	pin "github.com/jbenet/go-ipfs/pin"
 	routing "github.com/jbenet/go-ipfs/routing"
 	dht "github.com/jbenet/go-ipfs/routing/dht"
-	u "github.com/jbenet/go-ipfs/util"
 	ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
+	ds2 "github.com/jbenet/go-ipfs/util/datastore2"
 	debugerror "github.com/jbenet/go-ipfs/util/debugerror"
-	"github.com/jbenet/go-ipfs/util/eventlog"
+	eventlog "github.com/jbenet/go-ipfs/util/eventlog"
 )
 
 const IpnsValidatorTag = "ipns"
@@ -52,7 +52,7 @@ type IpfsNode struct {
 	Peerstore peer.Peerstore
 
 	// the local datastore
-	Datastore u.ThreadSafeDatastoreCloser
+	Datastore ds2.ThreadSafeDatastoreCloser
 
 	// the network message stream
 	Network inet.Network

core/datastore.go

@@ -10,10 +10,11 @@ import (
 
 	config "github.com/jbenet/go-ipfs/config"
 	u "github.com/jbenet/go-ipfs/util"
+	ds2 "github.com/jbenet/go-ipfs/util/datastore2"
 	"github.com/jbenet/go-ipfs/util/debugerror"
 )
 
-func makeDatastore(cfg config.Datastore) (u.ThreadSafeDatastoreCloser, error) {
+func makeDatastore(cfg config.Datastore) (ds2.ThreadSafeDatastoreCloser, error) {
 	if len(cfg.Type) == 0 {
 		return nil, debugerror.Errorf("config datastore.type required")
 	}
@@ -23,7 +24,7 @@ func makeDatastore(cfg config.Datastore) (u.ThreadSafeDatastoreCloser, error) {
 		return makeLevelDBDatastore(cfg)
 
 	case "memory":
-		return u.CloserWrap(syncds.MutexWrap(ds.NewMapDatastore())), nil
+		return ds2.CloserWrap(syncds.MutexWrap(ds.NewMapDatastore())), nil
 
 	case "fs":
 		log.Warning("using fs.Datastore at .datastore for testing.")
@@ -32,13 +33,13 @@ func makeDatastore(cfg config.Datastore) (u.ThreadSafeDatastoreCloser, error) {
 			return nil, err
 		}
 		ktd := ktds.Wrap(d, u.B58KeyConverter)
-		return u.CloserWrap(syncds.MutexWrap(ktd)), nil
+		return ds2.CloserWrap(syncds.MutexWrap(ktd)), nil
 	}
 
 	return nil, debugerror.Errorf("Unknown datastore type: %s", cfg.Type)
 }
 
-func makeLevelDBDatastore(cfg config.Datastore) (u.ThreadSafeDatastoreCloser, error) {
+func makeLevelDBDatastore(cfg config.Datastore) (ds2.ThreadSafeDatastoreCloser, error) {
 	if len(cfg.Path) == 0 {
 		return nil, debugerror.Errorf("config datastore.path required for leveldb")
 	}

core/mock.go

@@ -12,7 +12,7 @@ import (
 	path "github.com/jbenet/go-ipfs/path"
 	peer "github.com/jbenet/go-ipfs/peer"
 	mdht "github.com/jbenet/go-ipfs/routing/mock"
-	"github.com/jbenet/go-ipfs/util"
+	ds2 "github.com/jbenet/go-ipfs/util/datastore2"
 )
 
 // NewMockNode constructs an IpfsNode for use in tests.
@@ -39,7 +39,7 @@ func NewMockNode() (*IpfsNode, error) {
 
 	// Temp Datastore
 	dstore := ds.NewMapDatastore()
-	nd.Datastore = util.CloserWrap(syncds.MutexWrap(dstore))
+	nd.Datastore = ds2.CloserWrap(syncds.MutexWrap(dstore))
 
 	// Routing
 	dht := mdht.NewMockRouter(nd.Identity, nd.Datastore)

exchange/bitswap/bitswap_test.go

@@ -76,7 +76,7 @@ func TestGetBlockFromPeerAfterPeerAnnounces(t *testing.T) {
 
 	hasBlock := g.Next()
 
-	if err := hasBlock.Blockstore.Put(block); err != nil {
+	if err := hasBlock.Blockstore().Put(block); err != nil {
 		t.Fatal(err)
 	}
 	if err := hasBlock.Exchange.HasBlock(context.Background(), block); err != nil {
@@ -135,7 +135,7 @@ func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {
 
 	first := instances[0]
 	for _, b := range blocks {
-		first.Blockstore.Put(b)
+		first.Blockstore().Put(b)
 		first.Exchange.HasBlock(context.Background(), b)
 		rs.Announce(first.Peer, b.Key())
 	}
@@ -158,7 +158,7 @@ func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {
 
 	for _, inst := range instances {
 		for _, b := range blocks {
-			if _, err := inst.Blockstore.Get(b.Key()); err != nil {
+			if _, err := inst.Blockstore().Get(b.Key()); err != nil {
 				t.Fatal(err)
 			}
 		}
@@ -166,7 +166,7 @@ func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {
 }
 
 func getOrFail(bitswap Instance, b *blocks.Block, t *testing.T, wg *sync.WaitGroup) {
-	if _, err := bitswap.Blockstore.Get(b.Key()); err != nil {
+	if _, err := bitswap.Blockstore().Get(b.Key()); err != nil {
 		_, err := bitswap.Exchange.GetBlock(context.Background(), b.Key())
 		if err != nil {
 			t.Fatal(err)
@@ -208,7 +208,7 @@ func TestSendToWantingPeer(t *testing.T) {
 	beta := bg.Next()
 	t.Logf("Peer %v announes availability  of %v\n", w.Peer, beta.Key())
 	ctx, _ = context.WithTimeout(context.Background(), timeout)
-	if err := w.Blockstore.Put(beta); err != nil {
+	if err := w.Blockstore().Put(beta); err != nil {
 		t.Fatal(err)
 	}
 	w.Exchange.HasBlock(ctx, beta)
@@ -221,7 +221,7 @@ func TestSendToWantingPeer(t *testing.T) {
 
 	t.Logf("%v announces availability of %v\n", o.Peer, alpha.Key())
 	ctx, _ = context.WithTimeout(context.Background(), timeout)
-	if err := o.Blockstore.Put(alpha); err != nil {
+	if err := o.Blockstore().Put(alpha); err != nil {
 		t.Fatal(err)
 	}
 	o.Exchange.HasBlock(ctx, alpha)
@@ -233,7 +233,7 @@ func TestSendToWantingPeer(t *testing.T) {
 	}
 
 	t.Logf("%v should now have %v\n", w.Peer, alpha.Key())
-	block, err := w.Blockstore.Get(alpha.Key())
+	block, err := w.Blockstore().Get(alpha.Key())
 	if err != nil {
 		t.Fatalf("Should not have received an error: %s", err)
 	}

exchange/bitswap/testutils.go

@@ -1,14 +1,18 @@
 package bitswap
 
 import (
-	"github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	"time"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
 	ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
 	ds_sync "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
-	"github.com/jbenet/go-ipfs/blocks/blockstore"
+	blockstore "github.com/jbenet/go-ipfs/blocks/blockstore"
-	"github.com/jbenet/go-ipfs/exchange"
+	exchange "github.com/jbenet/go-ipfs/exchange"
 	tn "github.com/jbenet/go-ipfs/exchange/bitswap/testnet"
-	"github.com/jbenet/go-ipfs/peer"
+	peer "github.com/jbenet/go-ipfs/peer"
-	"github.com/jbenet/go-ipfs/routing/mock"
+	mock "github.com/jbenet/go-ipfs/routing/mock"
+	datastore2 "github.com/jbenet/go-ipfs/util/datastore2"
+	delay "github.com/jbenet/go-ipfs/util/delay"
 )
 
 func NewSessionGenerator(
@@ -45,7 +49,17 @@ func (g *SessionGenerator) Instances(n int) []Instance {
 type Instance struct {
 	Peer       peer.Peer
 	Exchange   exchange.Interface
-	Blockstore blockstore.Blockstore
+	blockstore blockstore.Blockstore
+
+	blockstoreDelay delay.D
+}
+
+func (i *Instance) Blockstore() blockstore.Blockstore {
+	return i.blockstore
+}
+
+func (i *Instance) SetBlockstoreLatency(t time.Duration) time.Duration {
+	return i.blockstoreDelay.Set(t)
 }
 
 // session creates a test bitswap session.
@@ -58,7 +72,9 @@ func session(net tn.Network, rs mock.RoutingServer, ps peer.Peerstore, id peer.I
 
 	adapter := net.Adapter(p)
 	htc := rs.Client(p)
-	bstore := blockstore.NewBlockstore(ds_sync.MutexWrap(ds.NewMapDatastore()))
+
+	bsdelay := delay.Fixed(0)
+	bstore := blockstore.NewBlockstore(ds_sync.MutexWrap(datastore2.WithDelay(ds.NewMapDatastore(), bsdelay)))
 
 	const alwaysSendToPeer = true
 	ctx := context.TODO()
@@ -66,8 +82,9 @@ func session(net tn.Network, rs mock.RoutingServer, ps peer.Peerstore, id peer.I
 	bs := New(ctx, p, adapter, htc, bstore, alwaysSendToPeer)
 
 	return Instance{
-		Peer:       p,
+		Peer:            p,
-		Exchange:   bs,
+		Exchange:        bs,
-		Blockstore: bstore,
+		blockstore:      bstore,
+		blockstoreDelay: bsdelay,
 	}
 }

importer/importer_test.go

@@ -9,10 +9,10 @@ import (
 	"os"
 	"testing"
 
-	"github.com/jbenet/go-ipfs/importer/chunk"
+	chunk "github.com/jbenet/go-ipfs/importer/chunk"
+	merkledag "github.com/jbenet/go-ipfs/merkledag"
 	uio "github.com/jbenet/go-ipfs/unixfs/io"
 	u "github.com/jbenet/go-ipfs/util"
-	testutil "github.com/jbenet/go-ipfs/util/testutil"
 )
 
 // NOTE:
@@ -91,7 +91,7 @@ func TestBuilderConsistency(t *testing.T) {
 	buf := new(bytes.Buffer)
 	io.CopyN(buf, u.NewTimeSeededRand(), int64(nbytes))
 	should := dup(buf.Bytes())
-	dagserv := testutil.GetDAGServ(t)
+	dagserv := merkledag.Mock(t)
 	nd, err := BuildDagFromReader(buf, dagserv, nil, chunk.DefaultSplitter)
 	if err != nil {
 		t.Fatal(err)

merkledag/merkledag_test.go

@@ -7,13 +7,10 @@ import (
 	"io/ioutil"
 	"testing"
 
-	bserv "github.com/jbenet/go-ipfs/blockservice"
+	blockservice "github.com/jbenet/go-ipfs/blockservice"
-	bs "github.com/jbenet/go-ipfs/exchange/bitswap"
-	tn "github.com/jbenet/go-ipfs/exchange/bitswap/testnet"
 	imp "github.com/jbenet/go-ipfs/importer"
 	chunk "github.com/jbenet/go-ipfs/importer/chunk"
 	. "github.com/jbenet/go-ipfs/merkledag"
-	"github.com/jbenet/go-ipfs/routing/mock"
 	uio "github.com/jbenet/go-ipfs/unixfs/io"
 	u "github.com/jbenet/go-ipfs/util"
 )
@@ -79,20 +76,8 @@ func makeTestDag(t *testing.T) *Node {
 }
 
 func TestBatchFetch(t *testing.T) {
-	net := tn.VirtualNetwork()
-	rs := mock.VirtualRoutingServer()
-	sg := bs.NewSessionGenerator(net, rs)
-
-	instances := sg.Instances(5)
-
-	var servs []*bserv.BlockService
 	var dagservs []DAGService
-	for _, i := range instances {
+	for _, bsi := range blockservice.Mocks(t, 5) {
-		bsi, err := bserv.New(i.Blockstore, i.Exchange)
-		if err != nil {
-			t.Fatal(err)
-		}
-		servs = append(servs, bsi)
 		dagservs = append(dagservs, NewDAGService(bsi))
 	}
 	t.Log("finished setup.")

merkledag/mock.go

@@ -0,0 +1,20 @@
+package merkledag
+
+import (
+	"testing"
+
+	ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
+	dssync "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
+	"github.com/jbenet/go-ipfs/blocks/blockstore"
+	bsrv "github.com/jbenet/go-ipfs/blockservice"
+	"github.com/jbenet/go-ipfs/exchange/offline"
+)
+
+func Mock(t testing.TB) DAGService {
+	bstore := blockstore.NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
+	bserv, err := bsrv.New(bstore, offline.Exchange(bstore))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return NewDAGService(bserv)
+}

routing/mock/routing.go

@@ -30,10 +30,6 @@ func NewMockRouter(local peer.Peer, dstore ds.Datastore) routing.IpfsRouting {
 	}
 }
 
-func (mr *MockRouter) SetRoutingServer(rs RoutingServer) {
-	mr.hashTable = rs
-}
-
 func (mr *MockRouter) PutValue(ctx context.Context, key u.Key, val []byte) error {
 	log.Debugf("PutValue: %s", key)
 	return mr.datastore.Put(key.DsKey(), val)
@@ -119,7 +115,8 @@ func (rs *hashTable) Announce(p peer.Peer, k u.Key) error {
 func (rs *hashTable) Providers(k u.Key) []peer.Peer {
 	rs.lock.RLock()
 	defer rs.lock.RUnlock()
-	ret := make([]peer.Peer, 0)
+
+	var ret []peer.Peer
 	peerset, ok := rs.providers[k]
 	if !ok {
 		return ret

util/datastore2/datastore_closer.go

@@ -1,4 +1,4 @@
-package util
+package datastore2
 
 import (
 	"io"

util/testutil/gen.go

@@ -2,28 +2,10 @@ package testutil
 
 import (
 	crand "crypto/rand"
-	"testing"
+	peer "github.com/jbenet/go-ipfs/peer"
-
-	dssync "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
-	"github.com/jbenet/go-ipfs/exchange/offline"
-	"github.com/jbenet/go-ipfs/peer"
-
-	ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
-	"github.com/jbenet/go-ipfs/blocks/blockstore"
-	bsrv "github.com/jbenet/go-ipfs/blockservice"
-	dag "github.com/jbenet/go-ipfs/merkledag"
 	u "github.com/jbenet/go-ipfs/util"
 )
 
-func GetDAGServ(t testing.TB) dag.DAGService {
-	bstore := blockstore.NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
-	bserv, err := bsrv.New(bstore, offline.Exchange(bstore))
-	if err != nil {
-		t.Fatal(err)
-	}
-	return dag.NewDAGService(bserv)
-}
-
 func RandPeer() peer.Peer {
 	id := make([]byte, 16)
 	crand.Read(id)

util/testutil/mock.go

@@ -1,9 +1,8 @@
 package testutil
 
 import (
-	"github.com/jbenet/go-ipfs/peer"
-
 	ic "github.com/jbenet/go-ipfs/crypto"
+	peer "github.com/jbenet/go-ipfs/peer"
 )
 
 func NewPeerWithKeyPair(sk ic.PrivKey, pk ic.PubKey) (peer.Peer, error) {