Fix test_BlockMetadataMatchedCoreAPI

tools/docker-network/tests/dockertestframework/accounts.go

@@ -48,7 +48,7 @@ func (d *DockerTestFramework) CheckAccountStatus(ctx context.Context, blkID iota
 
 	// Check the indexer
 	if len(checkIndexer) > 0 && checkIndexer[0] {
-		indexerClt, err := d.defaultWallet.Client.Indexer(ctx)
+		indexerClt, err := clt.Indexer(ctx)
 		require.NoError(d.Testing, err)
 
 		_, _, _, err = indexerClt.Account(ctx, accountAddress)
@@ -75,7 +75,9 @@ func (i *ImplicitAccount) OutputData() *mock.OutputData {
 
 // CreateImplicitAccount requests faucet funds and creates an implicit account. It already wait until the transaction is committed and the created account is useable.
 func (d *DockerTestFramework) CreateImplicitAccount(ctx context.Context, name string) *ImplicitAccount {
-	wallet := mock.NewWallet(d.Testing, name, d.defaultWallet.Client, &DockerWalletClock{client: d.defaultWallet.Client})
+	clt := d.defaultWallet.Client
+
+	wallet := mock.NewWallet(d.Testing, name, clt, &DockerWalletClock{client: clt})
 	outputData := d.RequestFaucetFunds(ctx, wallet, iotago.AddressImplicitAccountCreation)
 
 	accountID := iotago.AccountIDFromOutputID(outputData.ID)

tools/docker-network/tests/dockertestframework/awaits.go

@@ -127,18 +127,21 @@ func (d *DockerTestFramework) AwaitEpochFinalized() {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	info, err := d.defaultWallet.Client.Info(ctx)
+	clt := d.defaultWallet.Client
+
+	info, err := clt.Info(ctx)
 	require.NoError(d.Testing, err)
 
-	currentEpoch := d.defaultWallet.Client.CommittedAPI().TimeProvider().EpochFromSlot(info.Status.LatestFinalizedSlot)
+	currentEpoch := clt.CommittedAPI().TimeProvider().EpochFromSlot(info.Status.LatestFinalizedSlot)
 
 	// await the start slot of the next epoch
-	d.AwaitFinalizedSlot(d.defaultWallet.Client.CommittedAPI().TimeProvider().EpochStart(currentEpoch+1), true)
+	d.AwaitFinalizedSlot(clt.CommittedAPI().TimeProvider().EpochStart(currentEpoch+1), true)
 }
 
 func (d *DockerTestFramework) AwaitAddressUnspentOutputAccepted(ctx context.Context, wallet *mock.Wallet, addr iotago.Address) (outputID iotago.OutputID, output iotago.Output, err error) {
 	indexerClt, err := wallet.Client.Indexer(ctx)
 	require.NoError(d.Testing, err)
+
 	addrBech := addr.Bech32(d.defaultWallet.Client.CommittedAPI().ProtocolParameters().Bech32HRP())
 
 	for t := time.Now(); time.Since(t) < d.optsWaitFor; time.Sleep(d.optsTick) {

tools/docker-network/tests/dockertestframework/blocks.go

@@ -87,8 +87,6 @@ func (d *DockerTestFramework) CreateNFTBlockFromInput(wallet *mock.Wallet, input
 }
 
 func (d *DockerTestFramework) SubmitBlock(ctx context.Context, blk *iotago.Block) {
-	clt := d.defaultWallet.Client
-
-	_, err := clt.SubmitBlock(ctx, blk)
+	_, err := d.defaultWallet.Client.SubmitBlock(ctx, blk)
 	require.NoError(d.Testing, err)
 }

tools/docker-network/tests/dockertestframework/framework_eventapi.go

@@ -26,12 +26,6 @@ func WithEventAPIWaitFor(waitFor time.Duration) options.Option[EventAPIDockerTes
 	}
 }
 
-func WithEventAPITick(tick time.Duration) options.Option[EventAPIDockerTestFramework] {
-	return func(d *EventAPIDockerTestFramework) {
-		d.optsTick = tick
-	}
-}
-
 type EventAPIDockerTestFramework struct {
 	Testing *testing.T
 
@@ -41,7 +35,6 @@ type EventAPIDockerTestFramework struct {
 	finishChan chan struct{}
 
 	optsWaitFor time.Duration
-	optsTick    time.Duration
 }
 
 func NewEventAPIDockerTestFramework(t *testing.T, dockerFramework *DockerTestFramework) *EventAPIDockerTestFramework {
@@ -51,7 +44,6 @@ func NewEventAPIDockerTestFramework(t *testing.T, dockerFramework *DockerTestFra
 		DefaultClient:   dockerFramework.defaultWallet.Client,
 		finishChan:      make(chan struct{}),
 		optsWaitFor:     3 * time.Minute,
-		optsTick:        5 * time.Second,
 	}
 }
 
@@ -69,26 +61,29 @@ func (e *EventAPIDockerTestFramework) ConnectEventAPIClient(ctx context.Context)
 }
 
 // SubmitDataBlockStream submits a stream of data blocks to the network for the given duration.
-func (e *EventAPIDockerTestFramework) SubmitDataBlockStream(wallet *mock.Wallet, duration time.Duration) {
+func (e *EventAPIDockerTestFramework) SubmitDataBlockStream(wallet *mock.Wallet, duration time.Duration, tick time.Duration, countPerTick int, blockSubmittedCallback func()) {
 	timer := time.NewTimer(duration)
 	defer timer.Stop()
 
-	ticker := time.NewTicker(e.optsTick)
+	ticker := time.NewTicker(tick)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-ticker.C:
-			for i := 0; i < 10; i++ {
-				e.dockerFramework.defaultWallet.CreateAndSubmitBasicBlock(context.TODO(), "tagged_data_block", mock.WithPayload(tpkg.RandTaggedData([]byte("tag"))))
+			for range countPerTick {
+				_, err := wallet.CreateAndSubmitBasicBlock(context.TODO(), "tagged_data_block", mock.WithPayload(tpkg.RandTaggedData([]byte("tag"))))
+				require.NoError(e.Testing, err)
+
+				blockSubmittedCallback()
 			}
 		case <-timer.C:
 			return
 		}
 	}
 }
 
-func (e *EventAPIDockerTestFramework) AssertBlockMetadataStateAcceptedBlocks(ctx context.Context, eventClt *nodeclient.EventAPIClient) {
+func (e *EventAPIDockerTestFramework) AssertBlockMetadataStateAcceptedBlocks(ctx context.Context, eventClt *nodeclient.EventAPIClient, receivedCallback func()) {
 	acceptedChan, subInfo := eventClt.BlockMetadataAcceptedBlocks()
 	require.Nil(e.Testing, subInfo.Error())
 
@@ -103,19 +98,24 @@ func (e *EventAPIDockerTestFramework) AssertBlockMetadataStateAcceptedBlocks(ctx
 			case blk := <-acceptedChan:
 				require.Equal(e.Testing, api.BlockStateAccepted, blk.BlockState, "Block %s is pending in BlockMetadataAccepted topic", blk.BlockID.ToHex())
 
-				resp, err := eventClt.Client.BlockMetadataByBlockID(ctx, blk.BlockID)
+				resp, err := eventClt.Client.BlockWithMetadataByBlockID(ctx, blk.BlockID)
 				require.NoError(e.Testing, err)
+
 				// accepted, confirmed are accepted
-				require.NotEqualf(e.Testing, api.BlockStatePending, resp.BlockState, "Block %s is pending in BlockMetadataAccepted topic", blk.BlockID.ToHex())
+				require.NotEqualf(e.Testing, api.BlockStatePending, resp.Metadata.BlockState, "Block %s is pending in BlockMetadataAccepted topic", blk.BlockID.ToHex())
 
+				if resp.Block.Body.Type() == iotago.BlockBodyTypeBasic && resp.Block.Body.(*iotago.BasicBlockBody).Payload.PayloadType() == iotago.PayloadTaggedData {
+					// only count the basic blocks with tagged data, ignore the validation and candidate blocks
+					receivedCallback()
+				}
 			case <-ctx.Done():
 				return
 			}
 		}
 	}()
 }
 
-func (e *EventAPIDockerTestFramework) AssertBlockMetadataStateConfirmedBlocks(ctx context.Context, eventClt *nodeclient.EventAPIClient) {
+func (e *EventAPIDockerTestFramework) AssertBlockMetadataStateConfirmedBlocks(ctx context.Context, eventClt *nodeclient.EventAPIClient, receivedCallback func()) {
 	acceptedChan, subInfo := eventClt.BlockMetadataConfirmedBlocks()
 	require.Nil(e.Testing, subInfo.Error())
 
@@ -130,11 +130,15 @@ func (e *EventAPIDockerTestFramework) AssertBlockMetadataStateConfirmedBlocks(ct
 			case blk := <-acceptedChan:
 				require.Equal(e.Testing, api.BlockStateConfirmed, blk.BlockState, "Block %s is pending in BlockMetadataConfirmed topic", blk.BlockID.ToHex())
 
-				resp, err := eventClt.Client.BlockMetadataByBlockID(ctx, blk.BlockID)
+				resp, err := eventClt.Client.BlockWithMetadataByBlockID(ctx, blk.BlockID)
 				require.NoError(e.Testing, err)
-				require.NotEqualf(e.Testing, api.BlockStatePending, resp.BlockState, "Block %s is pending in BlockMetadataConfirmed endpoint", blk.BlockID.ToHex())
-				require.NotEqualf(e.Testing, api.BlockStateAccepted, resp.BlockState, "Block %s is accepted in BlockMetadataConfirmed endpoint", blk.BlockID.ToHex())
+				require.NotEqualf(e.Testing, api.BlockStatePending, resp.Metadata.BlockState, "Block %s is pending in BlockMetadataConfirmed endpoint", blk.BlockID.ToHex())
+				require.NotEqualf(e.Testing, api.BlockStateAccepted, resp.Metadata.BlockState, "Block %s is accepted in BlockMetadataConfirmed endpoint", blk.BlockID.ToHex())
 
+				if resp.Block.Body.Type() == iotago.BlockBodyTypeBasic && resp.Block.Body.(*iotago.BasicBlockBody).Payload.PayloadType() == iotago.PayloadTaggedData {
+					// only count the basic blocks with tagged data, ignore the validation and candidate blocks
+					receivedCallback()
+				}
 			case <-ctx.Done():
 				return
 			}
@@ -611,15 +615,15 @@ func (e *EventAPIDockerTestFramework) assertOutputMetadataTopics(ctx context.Con
 	}
 }
 
-func (e *EventAPIDockerTestFramework) AwaitEventAPITopics(t *testing.T, cancleFunc context.CancelFunc, numOfTopics int) error {
+func (e *EventAPIDockerTestFramework) AwaitEventAPITopics(t *testing.T, cancelFunc context.CancelFunc, numOfTopics int) error {
 	counter := 0
 	timer := time.NewTimer(e.optsWaitFor)
 	defer timer.Stop()
 
 	for {
 		select {
 		case <-timer.C:
-			cancleFunc()
+			cancelFunc()
 			return ierrors.New("Timeout, did not receive signals from all topics")
 		case <-e.finishChan:
 			counter++

tools/docker-network/tests/eventapi_test.go

@@ -5,6 +5,7 @@ package tests
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -36,7 +37,7 @@ func Test_MQTTTopics(t *testing.T) {
 	e := dockertestframework.NewEventAPIDockerTestFramework(t, d)
 
 	// prepare accounts to speed up tests
-	accounts := d.CreateAccountsFromFaucet(context.Background(), 2, "account-1", "account-2")
+	accounts := d.CreateAccountsFromFaucet(context.Background(), 5)
 
 	type test struct {
 		name     string
@@ -73,17 +74,17 @@ func Test_MQTTTopics(t *testing.T) {
 		{
 			name:     "Test_FoundryTransactionBlocks",
 			testFunc: test_FoundryTransactionBlocks,
-			account:  nil,
+			account:  accounts[2],
 		},
 		{
 			name:     "Test_NFTTransactionBlocks",
 			testFunc: test_NFTTransactionBlocks,
-			account:  nil,
+			account:  accounts[3],
 		},
 		{
 			name:     "Test_BlockMetadataMatchedCoreAPI",
 			testFunc: test_BlockMetadataMatchedCoreAPI,
-			account:  nil,
+			account:  accounts[4],
 		},
 	}
 
@@ -339,14 +340,13 @@ func test_AccountTransactionBlocks(t *testing.T, e *dockertestframework.EventAPI
 	}
 }
 
-func test_FoundryTransactionBlocks(t *testing.T, e *dockertestframework.EventAPIDockerTestFramework, _ *mock.AccountWithWallet) {
+func test_FoundryTransactionBlocks(t *testing.T, e *dockertestframework.EventAPIDockerTestFramework, account *mock.AccountWithWallet) {
 	// get event API client ready
 	ctx, cancel := context.WithCancel(context.Background())
 	eventClt := e.ConnectEventAPIClient(ctx)
 	defer eventClt.Close()
 
 	{
-		account := e.DockerTestFramework().CreateAccountFromFaucet("account-foundry")
 		fundsOutputData := e.DockerTestFramework().RequestFaucetFunds(ctx, account.Wallet(), iotago.AddressEd25519)
 
 		// prepare foundry output block
@@ -404,14 +404,13 @@ func test_FoundryTransactionBlocks(t *testing.T, e *dockertestframework.EventAPI
 	}
 }
 
-func test_NFTTransactionBlocks(t *testing.T, e *dockertestframework.EventAPIDockerTestFramework, _ *mock.AccountWithWallet) {
+func test_NFTTransactionBlocks(t *testing.T, e *dockertestframework.EventAPIDockerTestFramework, account *mock.AccountWithWallet) {
 	// get event API client ready
 	ctx, cancel := context.WithCancel(context.Background())
 	eventClt := e.ConnectEventAPIClient(ctx)
 	defer eventClt.Close()
 
 	{
-		account := e.DockerTestFramework().CreateAccountFromFaucet("account-nft")
 		fundsOutputData := e.DockerTestFramework().RequestFaucetFunds(ctx, account.Wallet(), iotago.AddressEd25519)
 
 		// prepare NFT output block
@@ -467,18 +466,28 @@ func test_NFTTransactionBlocks(t *testing.T, e *dockertestframework.EventAPIDock
 	}
 }
 
-func test_BlockMetadataMatchedCoreAPI(t *testing.T, e *dockertestframework.EventAPIDockerTestFramework, _ *mock.AccountWithWallet) {
+func test_BlockMetadataMatchedCoreAPI(t *testing.T, e *dockertestframework.EventAPIDockerTestFramework, account *mock.AccountWithWallet) {
 	// get event API client ready
 	ctx, cancel := context.WithCancel(context.Background())
 	eventClt := e.ConnectEventAPIClient(ctx)
 	defer eventClt.Close()
 
-	{
-		account := e.DockerTestFramework().CreateAccountFromFaucet("account-block-metadata")
+	receivedAcceptedCounter := atomic.Int64{}
+	receivedConfirmedCounter := atomic.Int64{}
+	sentCounter := atomic.Int64{}
 
+	{
 		assertions := []func(){
-			func() { e.AssertBlockMetadataStateAcceptedBlocks(ctx, eventClt) },
-			func() { e.AssertBlockMetadataStateConfirmedBlocks(ctx, eventClt) },
+			func() {
+				e.AssertBlockMetadataStateAcceptedBlocks(ctx, eventClt, func() {
+					receivedAcceptedCounter.Add(1)
+				})
+			},
+			func() {
+				e.AssertBlockMetadataStateConfirmedBlocks(ctx, eventClt, func() {
+					receivedConfirmedCounter.Add(1)
+				})
+			},
 		}
 
 		totalTopics := len(assertions)
@@ -491,8 +500,20 @@ func test_BlockMetadataMatchedCoreAPI(t *testing.T, e *dockertestframework.Event
 		require.NoError(t, err)
 
 		// issue blocks
-		e.SubmitDataBlockStream(account.Wallet(), 5*time.Minute)
+		fmt.Println("Submitting blocks for 30s...")
+		e.SubmitDataBlockStream(account.Wallet(), 30*time.Second, 1*time.Second, 10, func() {
+			sentCounter.Add(1)
+		})
 
+		// wait until all topics receives all expected objects
+		fmt.Println("Waiting for receiving additional blocks for 5s...")
+		time.Sleep(15 * time.Second)
+
+		// cancel listening
 		cancel()
+
+		// check if we received all expected objects
+		require.Equal(t, sentCounter.Load(), receivedAcceptedCounter.Load(), "receivedAcceptedCounter != sentCounter")
+		require.Equal(t, sentCounter.Load(), receivedConfirmedCounter.Load(), "receivedConfirmedCounter != sentCounter")
 	}
 }