test: fetching membership logic

waku/v2/protocol/rln/group_manager/dynamic/dynamic.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -40,7 +41,11 @@ type DynamicGroupManager struct {
 
 	membershipContractAddress common.Address
 	ethClientAddress          string
-	ethClient                 *ethclient.Client
+	ethClient                 interface {
+		BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error)
+		SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error)
+		Close()
+	}
 
 	lastBlockProcessed uint64
 

waku/v2/protocol/rln/group_manager/dynamic/mock_blockchain.go

@@ -0,0 +1,79 @@
+package dynamic
+
+import (
+	"math/big"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+)
+
+type MockBlockChain struct {
+	Blocks map[int64]*MockBlock `json:"blocks"`
+}
+
+type MockBlock []MockEvent
+
+func containsEntry[T common.Hash | common.Address](topics []T, topicA T) bool {
+	for _, topic := range topics {
+		if topic == topicA {
+			return true
+		}
+	}
+	return false
+}
+
+func Topic(topic string) common.Hash {
+	return crypto.Keccak256Hash([]byte(topic))
+}
+func (b MockBlock) getLogs(blockNum uint64, addrs []common.Address, topicA []common.Hash) (txLogs []types.Log) {
+	for ind, event := range b {
+		txLog := event.GetLog()
+		if containsEntry(addrs, txLog.Address) && (len(topicA) == 0 || containsEntry(topicA, txLog.Topics[0])) {
+			txLog.BlockNumber = blockNum
+			txLog.Index = uint(ind)
+			txLogs = append(txLogs, txLog)
+		}
+	}
+	return
+}
+
+type MockEvent struct {
+	Address common.Address `json:"address"`
+	Topics  []string       `json:"topics"`
+	Txhash  common.Hash    `json:"txhash"`
+	Data    []string       `json:"data"`
+}
+
+func (e MockEvent) GetLog() types.Log {
+	topics := []common.Hash{Topic(e.Topics[0])}
+	for _, topic := range e.Topics[1:] {
+		topics = append(topics, parseData(topic))
+	}
+	//
+	var data []byte
+	for _, entry := range e.Data {
+		data = append(data, parseData(entry).Bytes()...)
+	}
+	return types.Log{
+		Address: e.Address,
+		Topics:  topics,
+		TxHash:  e.Txhash,
+		Data:    data,
+	}
+}
+
+func parseData(data string) common.Hash {
+	splits := strings.Split(data, ":")
+	switch splits[0] {
+	case "bigint":
+		bigInt, ok := new(big.Int).SetString(splits[1], 10)
+		if !ok {
+			panic("invalid big int")
+		}
+		return common.BytesToHash(bigInt.Bytes())
+	default:
+		panic("invalid data type")
+	}
+}

waku/v2/protocol/rln/group_manager/dynamic/mock_client.go

@@ -0,0 +1,115 @@
+package dynamic
+
+import (
+	"context"
+	"encoding/json"
+	"io/ioutil"
+	"math/big"
+	"sort"
+	"testing"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+)
+
+type ErrCount struct {
+	err   error
+	count int
+}
+
+type MockClient struct {
+	ethclient.Client
+	blockChain     MockBlockChain
+	latestBlockNum int64
+	errOnBlock     map[int64]*ErrCount
+}
+
+func (c *MockClient) SetLatestBlockNumber(num int64) {
+	c.latestBlockNum = num
+}
+
+func (c MockClient) BlockByNumber(ctx context.Context, number *big.Int) (*types.Block, error) {
+	return types.NewBlock(&types.Header{Number: big.NewInt(c.latestBlockNum)}, nil, nil, nil, nil), nil
+}
+func NewMockClient(t *testing.T, blockFile string) *MockClient {
+	blockChain := MockBlockChain{}
+	data, err := ioutil.ReadFile(blockFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := json.Unmarshal(data, &blockChain); err != nil {
+		t.Fatal(err)
+	}
+	return &MockClient{blockChain: blockChain, errOnBlock: map[int64]*ErrCount{}}
+}
+
+func (client *MockClient) SetErrorOnBlock(blockNum int64, err error, count int) {
+	client.errOnBlock[blockNum] = &ErrCount{err: err, count: count}
+}
+
+func (c MockClient) getFromAndToRange(query ethereum.FilterQuery) (int64, int64) {
+	var fromBlock int64
+	if query.FromBlock == nil {
+		fromBlock = 0
+	} else {
+		fromBlock = query.FromBlock.Int64()
+	}
+
+	var toBlock int64
+	if query.ToBlock == nil {
+		toBlock = 0
+	} else {
+		toBlock = query.ToBlock.Int64()
+	}
+	return fromBlock, toBlock
+}
+func (c MockClient) FilterLogs(ctx context.Context, query ethereum.FilterQuery) (allTxLogs []types.Log, err error) {
+	fromBlock, toBlock := c.getFromAndToRange(query)
+	for block, details := range c.blockChain.Blocks {
+		if block >= fromBlock && block <= toBlock {
+			if txLogs := details.getLogs(uint64(block), query.Addresses, query.Topics[0]); len(txLogs) != 0 {
+				allTxLogs = append(allTxLogs, txLogs...)
+			}
+			if errCount, ok := c.errOnBlock[block]; ok && errCount.count != 0 {
+				errCount.count--
+				return nil, errCount.err
+			}
+		}
+	}
+	sort.Slice(allTxLogs, func(i, j int) bool {
+		return allTxLogs[i].BlockNumber < allTxLogs[j].BlockNumber ||
+			(allTxLogs[i].BlockNumber == allTxLogs[j].BlockNumber && allTxLogs[i].Index < allTxLogs[j].Index)
+	})
+	return allTxLogs, nil
+}
+
+func (c *MockClient) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
+	for {
+		next := c.latestBlockNum + 1
+		if c.blockChain.Blocks[next] != nil {
+			ch <- &types.Header{Number: big.NewInt(next)}
+			c.latestBlockNum = next
+		} else {
+			break
+		}
+	}
+	return testNoopSub{}, nil
+}
+
+type testNoopSub struct {
+}
+
+func (testNoopSub) Unsubscribe() {
+
+}
+
+// Err returns the subscription error channel. The error channel receives
+// a value if there is an issue with the subscription (e.g. the network connection
+// delivering the events has been closed). Only one value will ever be sent.
+// The error channel is closed by Unsubscribe.
+func (testNoopSub) Err() <-chan error {
+	ch := make(chan error)
+	close(ch)
+	return ch
+}

waku/v2/protocol/rln/group_manager/dynamic/web3.go

@@ -56,24 +56,21 @@ func (gm *DynamicGroupManager) HandleGroupUpdates(ctx context.Context, handler R
 }
 
 func (gm *DynamicGroupManager) loadOldEvents(ctx context.Context, rlnContract *contracts.RLN, fromBlock, toBlock uint64, handler RegistrationEventHandler) error {
-	var results []*contracts.RLNMemberRegistered
 	for ; fromBlock+maxBatchSize < toBlock; fromBlock += maxBatchSize + 1 { // check if the end of the batch is within the toBlock range
 		events, err := gm.getEvents(ctx, fromBlock, fromBlock+maxBatchSize)
 		if err != nil {
 			return err
 		}
-		results = append(results, events...)
+		handler(gm, events)
 	}
 
 	//
 	events, err := gm.getEvents(ctx, fromBlock, toBlock)
 	if err != nil {
 		return err
 	}
-	results = append(results, events...)
-	//
 	// process all the fetched events
-	return handler(gm, results)
+	return handler(gm, events)
 }
 
 func (gm *DynamicGroupManager) watchNewEvents(ctx context.Context, fromBlock uint64, handler RegistrationEventHandler, errCh chan<- error) {

waku/v2/protocol/rln/group_manager/dynamic/web3_test.go

@@ -0,0 +1,62 @@
+package dynamic
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/waku-org/go-waku/waku/v2/protocol/rln/contracts"
+	"github.com/waku-org/go-waku/waku/v2/utils"
+	"github.com/waku-org/go-zerokit-rln/rln"
+)
+
+var NULL_ADDR common.Address = common.HexToAddress("0x0000000000000000000000000000000000000000")
+
+func TestFetchingLogic(t *testing.T) {
+	client := NewMockClient(t, "web3_test.json")
+
+	rlnContract, err := contracts.NewRLN(NULL_ADDR, client)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rlnInstance, err := rln.NewRLN()
+	if err != nil {
+		t.Fatal(err)
+	}
+	dgm := DynamicGroupManager{
+		rlnContract: rlnContract,
+		rln:         rlnInstance,
+		log:         utils.Logger(),
+		ethClient:   client,
+	}
+
+	counts := []int{}
+	mockFn := func(_ *DynamicGroupManager, events []*contracts.RLNMemberRegistered) error {
+		counts = append(counts, len(events))
+		return nil
+	}
+	// check if more than 10k error is handled or not.
+	client.SetErrorOnBlock(5007, fmt.Errorf("query returned more than 10000 results"), 2)
+	client.SetLatestBlockNumber(10010)
+	dgm.HandleGroupUpdates(context.TODO(), mockFn)
+	//
+	time.Sleep(time.Second)
+	// check whether all the events are fetched or not.
+	if !sameArr(counts, []int{1, 3, 2, 1, 1}) {
+		t.Fatal("wrong no of events fetched per cycle", counts)
+	}
+}
+
+func sameArr(a, b []int) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i, v := range a {
+		if v != b[i] {
+			return false
+		}
+	}
+	return true
+}