Merge pull request #5380 from filecoin-project/refactor/deals-api-test

refactor deals API tests

api/test/deals.go

@@ -8,103 +8,40 @@ import (
 	"math/rand"
 	"os"
 	"path/filepath"
-	"sync/atomic"
 	"testing"
 	"time"
 
-	"github.com/filecoin-project/go-state-types/abi"
-
-	"github.com/stretchr/testify/require"
-
 	"github.com/ipfs/go-cid"
 	files "github.com/ipfs/go-ipfs-files"
 	"github.com/ipld/go-car"
+	"github.com/stretchr/testify/require"
 
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
+	"github.com/filecoin-project/lotus/chain/types"
 	sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
+	"github.com/filecoin-project/lotus/node/impl"
+	ipld "github.com/ipfs/go-ipld-format"
 	dag "github.com/ipfs/go-merkledag"
 	dstest "github.com/ipfs/go-merkledag/test"
 	unixfile "github.com/ipfs/go-unixfs/file"
-
-	"github.com/filecoin-project/lotus/chain/types"
-	"github.com/filecoin-project/lotus/node/impl"
-	ipld "github.com/ipfs/go-ipld-format"
 )
 
 func TestDealFlow(t *testing.T, b APIBuilder, blocktime time.Duration, carExport, fastRet bool, startEpoch abi.ChainEpoch) {
+	s := setupOneClientOneMiner(t, b, blocktime)
+	defer s.blockMiner.Stop()
 
-	ctx := context.Background()
+	MakeDeal(t, s.ctx, 6, s.client, s.miner, carExport, fastRet, startEpoch)
-	n, sn := b(t, OneFull, OneMiner)
-	client := n[0].FullNode.(*impl.FullNodeAPI)
-	miner := sn[0]
-
-	addrinfo, err := client.NetAddrsListen(ctx)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if err := miner.NetConnect(ctx, addrinfo); err != nil {
-		t.Fatal(err)
-	}
-	time.Sleep(time.Second)
-
-	mine := int64(1)
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		for atomic.LoadInt64(&mine) == 1 {
-			time.Sleep(blocktime)
-			if err := sn[0].MineOne(ctx, MineNext); err != nil {
-				t.Error(err)
-			}
-		}
-	}()
-
-	MakeDeal(t, ctx, 6, client, miner, carExport, fastRet, startEpoch)
-
-	atomic.AddInt64(&mine, -1)
-	fmt.Println("shutting down mining")
-	<-done
 }
 
 func TestDoubleDealFlow(t *testing.T, b APIBuilder, blocktime time.Duration, startEpoch abi.ChainEpoch) {
+	s := setupOneClientOneMiner(t, b, blocktime)
+	defer s.blockMiner.Stop()
 
-	ctx := context.Background()
+	MakeDeal(t, s.ctx, 6, s.client, s.miner, false, false, startEpoch)
-	n, sn := b(t, OneFull, OneMiner)
+	MakeDeal(t, s.ctx, 7, s.client, s.miner, false, false, startEpoch)
-	client := n[0].FullNode.(*impl.FullNodeAPI)
-	miner := sn[0]
-
-	addrinfo, err := client.NetAddrsListen(ctx)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if err := miner.NetConnect(ctx, addrinfo); err != nil {
-		t.Fatal(err)
-	}
-	time.Sleep(time.Second)
-
-	mine := int64(1)
-	done := make(chan struct{})
-
-	go func() {
-		defer close(done)
-		for atomic.LoadInt64(&mine) == 1 {
-			time.Sleep(blocktime)
-			if err := sn[0].MineOne(ctx, MineNext); err != nil {
-				t.Error(err)
-			}
-		}
-	}()
-
-	MakeDeal(t, ctx, 6, client, miner, false, false, startEpoch)
-	MakeDeal(t, ctx, 7, client, miner, false, false, startEpoch)
-
-	atomic.AddInt64(&mine, -1)
-	fmt.Println("shutting down mining")
-	<-done
 }
 
 func MakeDeal(t *testing.T, ctx context.Context, rseed int, client api.FullNode, miner TestStorageNode, carExport, fastRet bool, startEpoch abi.ChainEpoch) {
@@ -152,95 +89,41 @@ func CreateClientFile(ctx context.Context, client api.FullNode, rseed int) (*api
 }
 
 func TestFastRetrievalDealFlow(t *testing.T, b APIBuilder, blocktime time.Duration, startEpoch abi.ChainEpoch) {
-
+	s := setupOneClientOneMiner(t, b, blocktime)
-	ctx := context.Background()
+	defer s.blockMiner.Stop()
-	n, sn := b(t, OneFull, OneMiner)
-	client := n[0].FullNode.(*impl.FullNodeAPI)
-	miner := sn[0]
-
-	addrinfo, err := client.NetAddrsListen(ctx)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if err := miner.NetConnect(ctx, addrinfo); err != nil {
-		t.Fatal(err)
-	}
-	time.Sleep(time.Second)
-
-	mine := int64(1)
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		for atomic.LoadInt64(&mine) == 1 {
-			time.Sleep(blocktime)
-			if err := sn[0].MineOne(ctx, MineNext); err != nil {
-				t.Error(err)
-			}
-		}
-	}()
 
 	data := make([]byte, 1600)
 	rand.New(rand.NewSource(int64(8))).Read(data)
 
 	r := bytes.NewReader(data)
-	fcid, err := client.ClientImportLocal(ctx, r)
+	fcid, err := s.client.ClientImportLocal(s.ctx, r)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	fmt.Println("FILE CID: ", fcid)
 
-	deal := startDeal(t, ctx, miner, client, fcid, true, startEpoch)
+	deal := startDeal(t, s.ctx, s.miner, s.client, fcid, true, startEpoch)
 
-	waitDealPublished(t, ctx, miner, deal)
+	waitDealPublished(t, s.ctx, s.miner, deal)
 	fmt.Println("deal published, retrieving")
 	// Retrieval
-	info, err := client.ClientGetDealInfo(ctx, *deal)
+	info, err := s.client.ClientGetDealInfo(s.ctx, *deal)
 	require.NoError(t, err)
 
-	testRetrieval(t, ctx, client, fcid, &info.PieceCID, false, data)
+	testRetrieval(t, s.ctx, s.client, fcid, &info.PieceCID, false, data)
-	atomic.AddInt64(&mine, -1)
-	fmt.Println("shutting down mining")
-	<-done
 }
 
 func TestSecondDealRetrieval(t *testing.T, b APIBuilder, blocktime time.Duration) {
-
+	s := setupOneClientOneMiner(t, b, blocktime)
-	ctx := context.Background()
+	defer s.blockMiner.Stop()
-	n, sn := b(t, OneFull, OneMiner)
-	client := n[0].FullNode.(*impl.FullNodeAPI)
-	miner := sn[0]
-
-	addrinfo, err := client.NetAddrsListen(ctx)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if err := miner.NetConnect(ctx, addrinfo); err != nil {
-		t.Fatal(err)
-	}
-	time.Sleep(time.Second)
-
-	mine := int64(1)
-	done := make(chan struct{})
-
-	go func() {
-		defer close(done)
-		for atomic.LoadInt64(&mine) == 1 {
-			time.Sleep(blocktime)
-			if err := sn[0].MineOne(ctx, MineNext); err != nil {
-				t.Error(err)
-			}
-		}
-	}()
 
 	{
 		data1 := make([]byte, 800)
 		rand.New(rand.NewSource(int64(3))).Read(data1)
 		r := bytes.NewReader(data1)
 
-		fcid1, err := client.ClientImportLocal(ctx, r)
+		fcid1, err := s.client.ClientImportLocal(s.ctx, r)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -249,35 +132,31 @@ func TestSecondDealRetrieval(t *testing.T, b APIBuilder, blocktime time.Duration
 		rand.New(rand.NewSource(int64(9))).Read(data2)
 		r2 := bytes.NewReader(data2)
 
-		fcid2, err := client.ClientImportLocal(ctx, r2)
+		fcid2, err := s.client.ClientImportLocal(s.ctx, r2)
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		deal1 := startDeal(t, ctx, miner, client, fcid1, true, 0)
+		deal1 := startDeal(t, s.ctx, s.miner, s.client, fcid1, true, 0)
 
 		// TODO: this sleep is only necessary because deals don't immediately get logged in the dealstore, we should fix this
 		time.Sleep(time.Second)
-		waitDealSealed(t, ctx, miner, client, deal1, true)
+		waitDealSealed(t, s.ctx, s.miner, s.client, deal1, true)
 
-		deal2 := startDeal(t, ctx, miner, client, fcid2, true, 0)
+		deal2 := startDeal(t, s.ctx, s.miner, s.client, fcid2, true, 0)
 
 		time.Sleep(time.Second)
-		waitDealSealed(t, ctx, miner, client, deal2, false)
+		waitDealSealed(t, s.ctx, s.miner, s.client, deal2, false)
 
 		// Retrieval
-		info, err := client.ClientGetDealInfo(ctx, *deal2)
+		info, err := s.client.ClientGetDealInfo(s.ctx, *deal2)
 		require.NoError(t, err)
 
-		rf, _ := miner.SectorsRefs(ctx)
+		rf, _ := s.miner.SectorsRefs(s.ctx)
 		fmt.Printf("refs: %+v\n", rf)
 
-		testRetrieval(t, ctx, client, fcid2, &info.PieceCID, false, data2)
+		testRetrieval(t, s.ctx, s.client, fcid2, &info.PieceCID, false, data2)
 	}
-
-	atomic.AddInt64(&mine, -1)
-	fmt.Println("shutting down mining")
-	<-done
 }
 
 func startDeal(t *testing.T, ctx context.Context, miner TestStorageNode, client api.FullNode, fcid cid.Cid, fastRet bool, startEpoch abi.ChainEpoch) *cid.Cid {
@@ -459,3 +338,40 @@ func extractCarData(t *testing.T, ctx context.Context, rdata []byte, rpath strin
 	}
 	return rdata
 }
+
+type dealsScaffold struct {
+	ctx        context.Context
+	client     *impl.FullNodeAPI
+	miner      TestStorageNode
+	blockMiner *BlockMiner
+}
+
+func setupOneClientOneMiner(t *testing.T, b APIBuilder, blocktime time.Duration) *dealsScaffold {
+	n, sn := b(t, OneFull, OneMiner)
+	client := n[0].FullNode.(*impl.FullNodeAPI)
+	miner := sn[0]
+	return connectAndStartMining(t, b, blocktime, client, miner)
+}
+
+func connectAndStartMining(t *testing.T, b APIBuilder, blocktime time.Duration, client *impl.FullNodeAPI, miner TestStorageNode) *dealsScaffold {
+	ctx := context.Background()
+	addrinfo, err := client.NetAddrsListen(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := miner.NetConnect(ctx, addrinfo); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(time.Second)
+
+	blockMiner := NewBlockMiner(ctx, t, miner, blocktime)
+	blockMiner.MineBlocks()
+
+	return &dealsScaffold{
+		ctx:        ctx,
+		client:     client,
+		miner:      miner,
+		blockMiner: blockMiner,
+	}
+}