Implement test for RPC newHeads notification (#2805)

* implement test for `eth_subscribe`

* fix variable name collision

cmd/rpcdaemon/commands/eth_subscribe_test.go

@@ -0,0 +1,57 @@
+package commands
+
+import (
+	"testing"
+
+	"github.com/ledgerwatch/erigon-lib/gointerfaces/sentry"
+	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/filters"
+	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/rpcdaemontest"
+	"github.com/ledgerwatch/erigon/cmd/rpcdaemon/services"
+	"github.com/ledgerwatch/erigon/common"
+	"github.com/ledgerwatch/erigon/core"
+	"github.com/ledgerwatch/erigon/core/types"
+	"github.com/ledgerwatch/erigon/eth/protocols/eth"
+	"github.com/ledgerwatch/erigon/rlp"
+	"github.com/ledgerwatch/erigon/turbo/stages"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEthSubscribe(t *testing.T) {
+	m, require := stages.Mock(t), require.New(t)
+	chain, err := core.GenerateChain(m.ChainConfig, m.Genesis, m.Engine, m.DB, 42, func(i int, b *core.BlockGen) {
+		b.SetCoinbase(common.Address{1})
+	}, false /* intemediateHashes */)
+	require.NoError(err)
+
+	b, err := rlp.EncodeToBytes(&eth.BlockHeadersPacket66{
+		RequestId:          1,
+		BlockHeadersPacket: chain.Headers,
+	})
+	require.NoError(err)
+
+	m.ReceiveWg.Add(1)
+	for _, err = range m.Send(&sentry.InboundMessage{Id: sentry.MessageId_BLOCK_HEADERS_66, Data: b, PeerId: m.PeerId}) {
+		require.NoError(err)
+	}
+	m.ReceiveWg.Wait() // Wait for all messages to be processed before we proceeed
+
+	ctx, conn := rpcdaemontest.CreateTestGrpcConn(t, m)
+	backend := services.NewRemoteBackend(conn)
+	ff := filters.New(ctx, backend, nil, nil)
+
+	newHeads := make(chan *types.Header)
+	defer close(newHeads)
+	id := ff.SubscribeNewHeads(newHeads)
+	defer ff.UnsubscribeHeads(id)
+
+	initialCycle := true
+	highestSeenHeader := chain.TopBlock.NumberU64()
+	if err := stages.StageLoopStep(m.Ctx, m.DB, m.Sync, highestSeenHeader, m.Notifications, initialCycle, m.UpdateHead, nil); err != nil {
+		t.Fatal(err)
+	}
+
+	for i := uint64(1); i <= highestSeenHeader; i++ {
+		header := <-newHeads
+		require.Equal(header.Number.Uint64(), i)
+	}
+}

cmd/rpcdaemon/rpcdaemontest/test_util.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 
 	"github.com/holiman/uint256"
+	"github.com/ledgerwatch/erigon-lib/gointerfaces/remote"
 	"github.com/ledgerwatch/erigon-lib/gointerfaces/txpool"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon/accounts/abi/bind"
@@ -219,6 +220,7 @@ func CreateTestGrpcConn(t *testing.T, m *stages.MockSentry) (context.Context, *g
 	ethashApi := apis[1].Service.(*ethash.API)
 	server := grpc.NewServer()
 
+	remote.RegisterETHBACKENDServer(server, privateapi.NewEthBackendServer(ctx, nil, m.Notifications.Events))
 	txpool.RegisterTxpoolServer(server, privateapi.NewTxPoolServer(ctx, m.TxPoolP2PServer.TxPool))
 	txpool.RegisterMiningServer(server, privateapi.NewMiningServer(ctx, &IsMiningMock{}, ethashApi))
 	listener := bufconn.Listen(1024 * 1024)