Subscriptions test (#687)

* test: add tests for subscriptions types

* test: add tests for subscriptions block_reader

* test: add tests for subscriptions transfer_reader

* test: add tests for subscriptions beat_reader and beat2_reader

* refactor: added txPool when init chain in subscriptions test

* test: add tests for subscriptions pending_tx

* refactor: adding a mock event to a tx output receipt

* test: add tests for subscriptions event_reader

* test: add api/subscriptions test

* fix: add sync between reading and setting the txObject executable field

* test: checking convertEvent return an error

* refactor: split reader tests cases into separate functions

* refactor: add comments

* test: checking error instead of error message since could be different

* refactor: remove lock in txObject

* test: discard output logs in block_reader_test

api/subscriptions/beat2_reader_test.go

@@ -0,0 +1,66 @@
+// Copyright (c) 2024 The VeChainThor developers
+
+// Distributed under the GNU Lesser General Public License v3.0 software license, see the accompanying
+// file LICENSE or <https://www.gnu.org/licenses/lgpl-3.0.html>
+
+package subscriptions
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/vechain/thor/v2/thor"
+)
+
+func TestBeat2Reader_Read(t *testing.T) {
+	// Arrange
+	repo, generatedBlocks, _ := initChain(t)
+	genesisBlk := generatedBlocks[0]
+	newBlock := generatedBlocks[1]
+
+	// Act
+	beatReader := newBeat2Reader(repo, genesisBlk.Header().ID())
+	res, ok, err := beatReader.Read()
+
+	// Assert
+	assert.NoError(t, err)
+	assert.True(t, ok)
+	if beatMsg, ok := res[0].(*Beat2Message); !ok {
+		t.Fatal("unexpected type")
+	} else {
+		assert.Equal(t, newBlock.Header().Number(), beatMsg.Number)
+		assert.Equal(t, newBlock.Header().ID(), beatMsg.ID)
+		assert.Equal(t, newBlock.Header().ParentID(), beatMsg.ParentID)
+		assert.Equal(t, newBlock.Header().Timestamp(), beatMsg.Timestamp)
+		assert.Equal(t, uint32(newBlock.Header().TxsFeatures()), beatMsg.TxsFeatures)
+	}
+}
+
+func TestBeat2Reader_Read_NoNewBlocksToRead(t *testing.T) {
+	// Arrange
+	repo, generatedBlocks, _ := initChain(t)
+	newBlock := generatedBlocks[1]
+
+	// Act
+	beatReader := newBeat2Reader(repo, newBlock.Header().ID())
+	res, ok, err := beatReader.Read()
+
+	// Assert
+	assert.NoError(t, err)
+	assert.False(t, ok)
+	assert.Empty(t, res)
+}
+
+func TestBeat2Reader_Read_ErrorWhenReadingBlocks(t *testing.T) {
+	// Arrange
+	repo, _, _ := initChain(t)
+
+	// Act
+	beatReader := newBeat2Reader(repo, thor.MustParseBytes32("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"))
+	res, ok, err := beatReader.Read()
+
+	// Assert
+	assert.Error(t, err)
+	assert.False(t, ok)
+	assert.Empty(t, res)
+}

api/subscriptions/beat_reader_test.go

@@ -0,0 +1,66 @@
+// Copyright (c) 2024 The VeChainThor developers
+
+// Distributed under the GNU Lesser General Public License v3.0 software license, see the accompanying
+// file LICENSE or <https://www.gnu.org/licenses/lgpl-3.0.html>
+
+package subscriptions
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/vechain/thor/v2/thor"
+)
+
+func TestBeatReader_Read(t *testing.T) {
+	// Arrange
+	repo, generatedBlocks, _ := initChain(t)
+	genesisBlk := generatedBlocks[0]
+	newBlock := generatedBlocks[1]
+
+	// Act
+	beatReader := newBeatReader(repo, genesisBlk.Header().ID())
+	res, ok, err := beatReader.Read()
+
+	// Assert
+	assert.NoError(t, err)
+	assert.True(t, ok)
+	if beatMsg, ok := res[0].(*BeatMessage); !ok {
+		t.Fatal("unexpected type")
+	} else {
+		assert.Equal(t, newBlock.Header().Number(), beatMsg.Number)
+		assert.Equal(t, newBlock.Header().ID(), beatMsg.ID)
+		assert.Equal(t, newBlock.Header().ParentID(), beatMsg.ParentID)
+		assert.Equal(t, newBlock.Header().Timestamp(), beatMsg.Timestamp)
+		assert.Equal(t, uint32(newBlock.Header().TxsFeatures()), beatMsg.TxsFeatures)
+	}
+}
+
+func TestBeatReader_Read_NoNewBlocksToRead(t *testing.T) {
+	// Arrange
+	repo, generatedBlocks, _ := initChain(t)
+	newBlock := generatedBlocks[1]
+
+	// Act
+	beatReader := newBeatReader(repo, newBlock.Header().ID())
+	res, ok, err := beatReader.Read()
+
+	// Assert
+	assert.NoError(t, err)
+	assert.False(t, ok)
+	assert.Empty(t, res)
+}
+
+func TestBeatReader_Read_ErrorWhenReadingBlocks(t *testing.T) {
+	// Arrange
+	repo, _, _ := initChain(t)
+
+	// Act
+	beatReader := newBeatReader(repo, thor.MustParseBytes32("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"))
+	res, ok, err := beatReader.Read()
+
+	// Assert
+	assert.Error(t, err)
+	assert.False(t, ok)
+	assert.Empty(t, res)
+}

api/subscriptions/block_reader_test.go

@@ -0,0 +1,148 @@
+// Copyright (c) 2024 The VeChainThor developers
+
+// Distributed under the GNU Lesser General Public License v3.0 software license, see the accompanying
+// file LICENSE or <https://www.gnu.org/licenses/lgpl-3.0.html>
+
+package subscriptions
+
+import (
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/inconshreveable/log15"
+	"github.com/stretchr/testify/assert"
+	"github.com/vechain/thor/v2/block"
+	"github.com/vechain/thor/v2/chain"
+	"github.com/vechain/thor/v2/genesis"
+	"github.com/vechain/thor/v2/muxdb"
+	"github.com/vechain/thor/v2/packer"
+	"github.com/vechain/thor/v2/state"
+	"github.com/vechain/thor/v2/thor"
+	"github.com/vechain/thor/v2/tx"
+	"github.com/vechain/thor/v2/txpool"
+)
+
+func init() {
+	log15.Root().SetHandler(log15.DiscardHandler())
+}
+
+func TestBlockReader_Read(t *testing.T) {
+	repo, generatedBlocks, _ := initChain(t)
+	genesisBlk := generatedBlocks[0]
+	newBlock := generatedBlocks[1]
+
+	// Test case 1: Successful read next blocks
+	br := newBlockReader(repo, genesisBlk.Header().ID())
+	res, ok, err := br.Read()
+
+	assert.NoError(t, err)
+	assert.True(t, ok)
+	if resBlock, ok := res[0].(*BlockMessage); !ok {
+		t.Fatal("unexpected type")
+	} else {
+		assert.Equal(t, newBlock.Header().Number(), resBlock.Number)
+		assert.Equal(t, newBlock.Header().ParentID(), resBlock.ParentID)
+	}
+
+	// Test case 2: There is no new block
+	br = newBlockReader(repo, newBlock.Header().ID())
+	res, ok, err = br.Read()
+
+	assert.NoError(t, err)
+	assert.False(t, ok)
+	assert.Empty(t, res)
+
+	// Test case 3: Error when reading blocks
+	br = newBlockReader(repo, thor.MustParseBytes32("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"))
+	res, ok, err = br.Read()
+
+	assert.Error(t, err)
+	assert.False(t, ok)
+	assert.Empty(t, res)
+}
+
+func initChain(t *testing.T) (*chain.Repository, []*block.Block, *txpool.TxPool) {
+	db := muxdb.NewMem()
+	stater := state.NewStater(db)
+	gene := genesis.NewDevnet()
+
+	b, _, _, err := gene.Build(stater)
+	if err != nil {
+		t.Fatal(err)
+	}
+	repo, _ := chain.NewRepository(db, b)
+
+	txPool := txpool.New(repo, stater, txpool.Options{
+		Limit:           100,
+		LimitPerAccount: 16,
+		MaxLifetime:     time.Hour,
+	})
+
+	addr := thor.BytesToAddress([]byte("to"))
+	cla := tx.NewClause(&addr).WithValue(big.NewInt(10000))
+	tr := new(tx.Builder).
+		ChainTag(repo.ChainTag()).
+		GasPriceCoef(1).
+		Expiration(10).
+		Gas(21000).
+		Nonce(1).
+		Clause(cla).
+		BlockRef(tx.NewBlockRef(0)).
+		Build()
+
+	sig, err := crypto.Sign(tr.SigningHash().Bytes(), genesis.DevAccounts()[0].PrivateKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tr = tr.WithSignature(sig)
+	packer := packer.New(repo, stater, genesis.DevAccounts()[0].Address, &genesis.DevAccounts()[0].Address, thor.NoFork)
+	sum, _ := repo.GetBlockSummary(b.Header().ID())
+	flow, err := packer.Schedule(sum, uint64(time.Now().Unix()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = flow.Adopt(tr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	blk, stage, receipts, err := flow.Pack(genesis.DevAccounts()[0].PrivateKey, 0, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := stage.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	insertMockOutputEvent(receipts)
+	if err := repo.AddBlock(blk, receipts, 0); err != nil {
+		t.Fatal(err)
+	}
+	if err := repo.SetBestBlockID(blk.Header().ID()); err != nil {
+		t.Fatal(err)
+	}
+	return repo, []*block.Block{b, blk}, txPool
+}
+
+// This is a helper function to forcly insert an event into the output receipts
+func insertMockOutputEvent(receipts tx.Receipts) {
+	oldReceipt := receipts[0]
+	events := make(tx.Events, 0)
+	events = append(events, &tx.Event{
+		Address: thor.BytesToAddress([]byte("to")),
+		Topics:  []thor.Bytes32{thor.BytesToBytes32([]byte("topic"))},
+		Data:    []byte("data"),
+	})
+	outputs := &tx.Output{
+		Transfers: oldReceipt.Outputs[0].Transfers,
+		Events:    events,
+	}
+	receipts[0] = &tx.Receipt{
+		Reverted: oldReceipt.Reverted,
+		GasUsed:  oldReceipt.GasUsed,
+		Outputs:  []*tx.Output{outputs},
+		GasPayer: oldReceipt.GasPayer,
+		Paid:     oldReceipt.Paid,
+		Reward:   oldReceipt.Reward,
+	}
+}

api/subscriptions/event_reader_test.go

@@ -0,0 +1,57 @@
+// Copyright (c) 2024 The VeChainThor developers
+
+// Distributed under the GNU Lesser General Public License v3.0 software license, see the accompanying
+// file LICENSE or <https://www.gnu.org/licenses/lgpl-3.0.html>
+
+package subscriptions
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/vechain/thor/v2/chain"
+)
+
+func TestEventReader_Read(t *testing.T) {
+	repo, generatedBlocks, _ := initChain(t)
+	genesisBlk := generatedBlocks[0]
+	newBlock := generatedBlocks[1]
+
+	er := &eventReader{
+		repo:        repo,
+		filter:      &EventFilter{},
+		blockReader: &mockBlockReaderWithError{},
+	}
+
+	// Test case 1: An error occurred while reading blocks
+	events, ok, err := er.Read()
+	assert.Error(t, err)
+	assert.Empty(t, events)
+	assert.False(t, ok)
+
+	// Test case 2: Events are available to read
+	er = newEventReader(repo, genesisBlk.Header().ID(), &EventFilter{})
+
+	events, ok, err = er.Read()
+
+	assert.NoError(t, err)
+	assert.True(t, ok)
+	var eventMessages []*EventMessage
+	for _, event := range events {
+		if msg, ok := event.(*EventMessage); ok {
+			eventMessages = append(eventMessages, msg)
+		} else {
+			t.Fatal("unexpected type")
+		}
+	}
+	assert.Equal(t, 1, len(eventMessages))
+	eventMsg := eventMessages[0]
+	assert.Equal(t, newBlock.Header().ID(), eventMsg.Meta.BlockID)
+	assert.Equal(t, newBlock.Header().Number(), eventMsg.Meta.BlockNumber)
+}
+
+type mockBlockReaderWithError struct{}
+
+func (m *mockBlockReaderWithError) Read() ([]*chain.ExtendedBlock, error) {
+	return nil, assert.AnError
+}