fix tests

core/exchange_test.go

@@ -73,7 +73,6 @@ func TestExchange_DoNotStoreHistoric(t *testing.T) {
 	fetcher, cctx := createCoreFetcher(t, cfg)
 	generateNonEmptyBlocks(t, ctx, fetcher, cfg, cctx)
 
-	require.NoError(t, fetcher.Start(ctx))
 	store, err := store.NewStore(store.DefaultParameters(), t.TempDir())
 	require.NoError(t, err)
 
@@ -208,11 +207,9 @@ func generateNonEmptyBlocks(
 	// generate several non-empty blocks
 	generateCtx, generateCtxCancel := context.WithCancel(context.Background())
 
-	require.NoError(t, fetcher.Start(ctx))
-	sub, err := fetcher.runSubscriber()
+	sub, err := fetcher.SubscribeNewBlockEvent(generateCtx)
 	require.NoError(t, err)
 	defer func() {
-		err = fetcher.Stop(ctx)
 		require.NoError(t, err)
 	}()
 

core/fetcher_no_race_test.go

@@ -25,10 +25,7 @@ func TestBlockFetcherHeaderValues(t *testing.T) {
 	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
-	err = fetcher.Start(ctx)
-	require.NoError(t, err)
-	// generate some blocks
-	newBlockChan, err := fetcher.runSubscriber()
+	newBlockChan, err := fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
 	// read once from channel to generate next block
 	var h int64
@@ -57,5 +54,4 @@ func TestBlockFetcherHeaderValues(t *testing.T) {
 	// compare ValidatorSet hash to the ValidatorsHash from first block height
 	hexBytes := valSet.Hash()
 	assert.Equal(t, nextBlock.ValidatorSet.Hash(), hexBytes)
-	require.NoError(t, fetcher.Stop(ctx))
 }

core/fetcher_test.go

@@ -19,10 +19,8 @@ func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
 	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
-	err = fetcher.Start(ctx)
-	require.NoError(t, err)
 	// generate some blocks
-	newBlockChan, err := fetcher.runSubscriber()
+	newBlockChan, err := fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
 
 	for i := 1; i < 3; i++ {
@@ -40,13 +38,11 @@ func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
 			require.NoError(t, ctx.Err())
 		}
 	}
-	require.NoError(t, fetcher.Stop(ctx))
 }
 
 // TestFetcher_Resubscription ensures that subscription will not stuck in case
 // gRPC server was stopped.
 func TestFetcher_Resubscription(t *testing.T) {
-	t.Skip()
 	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
 	t.Cleanup(cancel)
 	// run new consensus node
@@ -58,12 +54,10 @@ func TestFetcher_Resubscription(t *testing.T) {
 	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
-	err = fetcher.Start(ctx)
-	require.NoError(t, err)
 
 	// subscribe to the channel to get new blocks
 	// and try to get one block
-	newBlockChan, err := fetcher.runSubscriber()
+	newBlockChan, err := fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
 	select {
 	case newBlockFromChan := <-newBlockChan:
@@ -99,7 +93,5 @@ func TestFetcher_Resubscription(t *testing.T) {
 	case <-ctx.Done():
 		t.Fatal("timeout waiting for block subscription")
 	}
-
 	require.NoError(t, tn.Stop())
-	require.NoError(t, fetcher.Stop(ctx))
 }

core/header_test.go

@@ -26,13 +26,11 @@ func TestMakeExtendedHeaderForEmptyBlock(t *testing.T) {
 	client := newTestClient(t, host, port)
 	fetcher, err := NewBlockFetcher(client)
 	require.NoError(t, err)
-	err = fetcher.Start(ctx)
+	sub, err := fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
-	sub, err := fetcher.runSubscriber()
-	require.NoError(t, err)
-	<-sub
+	dataBlock := <-sub
 
-	height := int64(1)
+	height := dataBlock.Header.Height
 	b, err := fetcher.GetBlock(ctx, height)
 	require.NoError(t, err)
 
@@ -46,7 +44,6 @@ func TestMakeExtendedHeaderForEmptyBlock(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.Equal(t, share.EmptyEDSRoots(), headerExt.DAH)
-	require.NoError(t, fetcher.Stop(ctx))
 }
 
 func TestMismatchedDataHash_ComputedRoot(t *testing.T) {

core/listener_test.go

@@ -90,8 +90,7 @@ func TestListenerWithWrongChainRPC(t *testing.T) {
 
 	// create Listener and start listening
 	cl := createListener(ctx, t, fetcher, ps0, eds, store, "wrong-chain-rpc")
-	require.NoError(t, cl.fetcher.Start(ctx))
-	sub, err := cl.fetcher.runSubscriber()
+	sub, err := cl.fetcher.SubscribeNewBlockEvent(ctx)
 	require.NoError(t, err)
 
 	assert.Panics(t, func() { cl.listen(ctx, sub) })

core/testing.go

@@ -104,7 +104,7 @@ func newTestClient(t *testing.T, ip, port string) *grpc.ClientConn {
 }
 
 // Network wraps `testnode.Context` allowing to manually stop all underlying connections.
-// TODO: remove after https://github.com/celestiaorg/celestia-app/issues/4304 is done.
+// TODO @vgonkivs: remove after https://github.com/celestiaorg/celestia-app/issues/4304 is done.
 type Network struct {
 	testnode.Context
 	config *testnode.Config