Merge pull request #6846 from The-K-R-O-K/AndriiDiachuk/6819-make-res…

…ponse-for-data-providers-consistent

[Access] Make response for data providers consistent.

engine/access/rest/websockets/data_providers/account_statuses_provider.go

@@ -117,8 +117,11 @@ func (p *AccountStatusesDataProvider) handleResponse() func(accountStatusesRespo
 			return status.Errorf(codes.Internal, "message index already incremented to %d", messageIndex.Value())
 		}
 
-		var response models.AccountStatusesResponse
-		response.Build(accountStatusesResponse, index)
+		var accountStatusesPayload models.AccountStatusesResponse
+		accountStatusesPayload.Build(accountStatusesResponse, index)
+
+		var response models.BaseDataProvidersResponse
+		response.Build(p.ID(), p.Topic(), &accountStatusesPayload)
 
 		p.send <- &response
 

engine/access/rest/websockets/data_providers/account_statuses_provider_test.go

@@ -139,25 +139,40 @@ func (s *AccountStatusesProviderSuite) subscribeAccountStatusesDataProviderTestC
 
 // requireAccountStatuses ensures that the received account statuses information matches the expected data.
 func (s *AccountStatusesProviderSuite) requireAccountStatuses(actual interface{}, expected interface{}) {
-	expectedResponse, ok := expected.(*models.AccountStatusesResponse)
-	require.True(s.T(), ok, "Expected *models.AccountStatusesResponse, got %T", expected)
+	expectedResponse, expectedResponsePayload := extractPayload[*models.AccountStatusesResponse](s.T(), expected)
+	actualResponse, actualResponsePayload := extractPayload[*models.AccountStatusesResponse](s.T(), actual)
 
-	actualResponse, ok := actual.(*models.AccountStatusesResponse)
-	require.True(s.T(), ok, "Expected *models.AccountStatusesResponse, got %T", actual)
+	require.Equal(s.T(), expectedResponsePayload.BlockID, actualResponsePayload.BlockID)
+	require.Equal(s.T(), len(expectedResponsePayload.AccountEvents), len(actualResponsePayload.AccountEvents))
+	require.Equal(s.T(), expectedResponsePayload.MessageIndex, actualResponsePayload.MessageIndex)
+	require.Equal(s.T(), expectedResponsePayload.Height, actualResponsePayload.Height)
+	require.Equal(s.T(), expectedResponse.Topic, actualResponse.Topic)
 
-	require.Equal(s.T(), expectedResponse.BlockID, actualResponse.BlockID)
-	require.Equal(s.T(), len(expectedResponse.AccountEvents), len(actualResponse.AccountEvents))
-	require.Equal(s.T(), expectedResponse.MessageIndex, actualResponse.MessageIndex)
-	require.Equal(s.T(), expectedResponse.Height, actualResponse.Height)
-
-	for key, expectedEvents := range expectedResponse.AccountEvents {
-		actualEvents, ok := actualResponse.AccountEvents[key]
+	for key, expectedEvents := range expectedResponsePayload.AccountEvents {
+		actualEvents, ok := actualResponsePayload.AccountEvents[key]
 		require.True(s.T(), ok, "Missing key in actual AccountEvents: %s", key)
 
 		s.Require().Equal(expectedEvents, actualEvents, "Mismatch for key: %s", key)
 	}
 }
 
+// expectedAccountStatusesResponses creates the expected responses for the provided events and backend responses.
+func (s *AccountStatusesProviderSuite) expectedAccountStatusesResponses(backendResponses []*backend.AccountStatusesResponse) []interface{} {
+	expectedResponses := make([]interface{}, len(backendResponses))
+
+	for i, resp := range backendResponses {
+		var expectedResponsePayload models.AccountStatusesResponse
+		expectedResponsePayload.Build(resp, uint64(i))
+
+		expectedResponses[i] = &models.BaseDataProvidersResponse{
+			Topic:   AccountStatusesTopic,
+			Payload: &expectedResponsePayload,
+		}
+	}
+
+	return expectedResponses
+}
+
 // TestAccountStatusesDataProvider_InvalidArguments tests the behavior of the account statuses data provider
 // when invalid arguments are provided. It verifies that appropriate errors are returned
 // for missing or conflicting arguments.
@@ -254,9 +269,10 @@ func (s *AccountStatusesProviderSuite) TestMessageIndexAccountStatusesProviderRe
 	var responses []*models.AccountStatusesResponse
 	for i := 0; i < accountStatusesCount; i++ {
 		res := <-send
-		accountStatusesRes, ok := res.(*models.AccountStatusesResponse)
-		s.Require().True(ok, "Expected *models.AccountStatusesResponse, got %T", res)
-		responses = append(responses, accountStatusesRes)
+
+		_, accStatusesResponsePayload := extractPayload[*models.AccountStatusesResponse](s.T(), res)
+
+		responses = append(responses, accStatusesResponsePayload)
 	}
 
 	// Wait for the provider goroutine to finish
@@ -289,17 +305,3 @@ func (s *AccountStatusesProviderSuite) backendAccountStatusesResponses(events []
 
 	return responses
 }
-
-// expectedAccountStatusesResponses creates the expected responses for the provided events and backend responses.
-func (s *AccountStatusesProviderSuite) expectedAccountStatusesResponses(backendResponses []*backend.AccountStatusesResponse) []interface{} {
-	expectedResponses := make([]interface{}, len(backendResponses))
-
-	for i, resp := range backendResponses {
-		var expectedResponse models.AccountStatusesResponse
-		expectedResponse.Build(resp, uint64(i))
-
-		expectedResponses[i] = &expectedResponse
-	}
-
-	return expectedResponses
-}

engine/access/rest/websockets/data_providers/block_digests_provider.go

@@ -66,9 +66,14 @@ func (p *BlockDigestsDataProvider) Run() error {
 			var block models.BlockDigest
 			block.Build(b)
 
-			return &models.BlockDigestMessageResponse{
-				Block: &block,
-			}, nil
+			var response models.BaseDataProvidersResponse
+			response.Build(
+				p.ID(),
+				p.Topic(),
+				&block,
+			)
+
+			return &response, nil
 		}),
 	)
 }

engine/access/rest/websockets/data_providers/block_digests_provider_test.go

@@ -6,7 +6,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/onflow/flow-go/engine/access/rest/common/parser"
@@ -28,27 +27,23 @@ func (s *BlockDigestsProviderSuite) SetupTest() {
 	s.BlocksProviderSuite.SetupTest()
 }
 
-// TestBlockDigestsDataProvider_InvalidArguments tests the behavior of the block digests data provider
-// when invalid arguments are provided. It verifies that appropriate errors are returned
-// for missing or conflicting arguments.
-// This test covers the test cases:
-// 1. Missing 'block_status' argument.
-// 2. Invalid 'block_status' argument.
-// 3. Providing both 'start_block_id' and 'start_block_height' simultaneously.
-func (s *BlockDigestsProviderSuite) TestBlockDigestsDataProvider_InvalidArguments() {
-	ctx := context.Background()
-	send := make(chan interface{})
-
-	topic := BlockDigestsTopic
-
-	for _, test := range s.invalidArgumentsTestCases() {
-		s.Run(test.name, func() {
-			provider, err := NewBlockDigestsDataProvider(ctx, s.log, s.api, "dummy-id", topic, test.arguments, send)
-			s.Require().Nil(provider)
-			s.Require().Error(err)
-			s.Require().Contains(err.Error(), test.expectedErrorMsg)
-		})
-	}
+// TestBlockDigestsDataProvider_HappyPath tests the behavior of the block digests data provider
+// when it is configured correctly and operating under normal conditions. It
+// validates that block digests are correctly streamed to the channel and ensures
+// no unexpected errors occur.
+func (s *BlockDigestsProviderSuite) TestBlockDigestsDataProvider_HappyPath() {
+	testHappyPath(
+		s.T(),
+		BlockDigestsTopic,
+		s.factory,
+		s.validBlockDigestsArgumentsTestCases(),
+		func(dataChan chan interface{}) {
+			for _, block := range s.blocks {
+				dataChan <- flow.NewBlockDigest(block.Header.ID(), block.Header.Height, block.Header.Timestamp)
+			}
+		},
+		s.requireBlockDigest,
+	)
 }
 
 // validBlockDigestsArgumentsTestCases defines test happy cases for block digests data providers.
@@ -61,7 +56,10 @@ func (s *BlockDigestsProviderSuite) validBlockDigestsArgumentsTestCases() []test
 		var block models.BlockDigest
 		block.Build(blockDigest)
 
-		expectedResponses[i] = &models.BlockDigestMessageResponse{Block: &block}
+		expectedResponses[i] = &models.BaseDataProvidersResponse{
+			Topic:   BlockDigestsTopic,
+			Payload: &block,
+		}
 	}
 
 	return []testType{
@@ -114,32 +112,34 @@ func (s *BlockDigestsProviderSuite) validBlockDigestsArgumentsTestCases() []test
 	}
 }
 
-// TestBlockDigestsDataProvider_HappyPath tests the behavior of the block digests data provider
-// when it is configured correctly and operating under normal conditions. It
-// validates that block digests are correctly streamed to the channel and ensures
-// no unexpected errors occur.
-func (s *BlockDigestsProviderSuite) TestBlockDigestsDataProvider_HappyPath() {
-	testHappyPath(
-		s.T(),
-		BlockDigestsTopic,
-		s.factory,
-		s.validBlockDigestsArgumentsTestCases(),
-		func(dataChan chan interface{}) {
-			for _, block := range s.blocks {
-				dataChan <- flow.NewBlockDigest(block.Header.ID(), block.Header.Height, block.Header.Timestamp)
-			}
-		},
-		s.requireBlockDigest,
-	)
+// requireBlockDigest ensures that the received block header information matches the expected data.
+func (s *BlocksProviderSuite) requireBlockDigest(actual interface{}, expected interface{}) {
+	expectedResponse, expectedResponsePayload := extractPayload[*models.BlockDigest](s.T(), expected)
+	actualResponse, actualResponsePayload := extractPayload[*models.BlockDigest](s.T(), actual)
+
+	s.Require().Equal(expectedResponse.Topic, actualResponse.Topic)
+	s.Require().Equal(expectedResponsePayload, actualResponsePayload)
 }
 
-// requireBlockHeaders ensures that the received block header information matches the expected data.
-func (s *BlocksProviderSuite) requireBlockDigest(actual interface{}, expected interface{}) {
-	actualResponse, ok := actual.(*models.BlockDigestMessageResponse)
-	require.True(s.T(), ok, "unexpected response type: %T", actual)
+// TestBlockDigestsDataProvider_InvalidArguments tests the behavior of the block digests data provider
+// when invalid arguments are provided. It verifies that appropriate errors are returned
+// for missing or conflicting arguments.
+// This test covers the test cases:
+// 1. Missing 'block_status' argument.
+// 2. Invalid 'block_status' argument.
+// 3. Providing both 'start_block_id' and 'start_block_height' simultaneously.
+func (s *BlockDigestsProviderSuite) TestBlockDigestsDataProvider_InvalidArguments() {
+	ctx := context.Background()
+	send := make(chan interface{})
 
-	expectedResponse, ok := expected.(*models.BlockDigestMessageResponse)
-	require.True(s.T(), ok, "unexpected response type: %T", expected)
+	topic := BlockDigestsTopic
 
-	s.Require().Equal(expectedResponse.Block, actualResponse.Block)
+	for _, test := range s.invalidArgumentsTestCases() {
+		s.Run(test.name, func() {
+			provider, err := NewBlockDigestsDataProvider(ctx, s.log, s.api, "dummy-id", topic, test.arguments, send)
+			s.Require().Nil(provider)
+			s.Require().Error(err)
+			s.Require().Contains(err.Error(), test.expectedErrorMsg)
+		})
+	}
 }

engine/access/rest/websockets/data_providers/block_headers_provider.go

@@ -67,9 +67,14 @@ func (p *BlockHeadersDataProvider) Run() error {
 			var header commonmodels.BlockHeader
 			header.Build(h)
 
-			return &models.BlockHeaderMessageResponse{
-				Header: &header,
-			}, nil
+			var response models.BaseDataProvidersResponse
+			response.Build(
+				p.ID(),
+				p.Topic(),
+				&header,
+			)
+
+			return &response, nil
 		}),
 	)
 }

engine/access/rest/websockets/data_providers/block_headers_provider_test.go

@@ -6,7 +6,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
 	commonmodels "github.com/onflow/flow-go/engine/access/rest/common/models"
@@ -29,27 +28,23 @@ func (s *BlockHeadersProviderSuite) SetupTest() {
 	s.BlocksProviderSuite.SetupTest()
 }
 
-// TestBlockHeadersDataProvider_InvalidArguments tests the behavior of the block headers data provider
-// when invalid arguments are provided. It verifies that appropriate errors are returned
-// for missing or conflicting arguments.
-// This test covers the test cases:
-// 1. Missing 'block_status' argument.
-// 2. Invalid 'block_status' argument.
-// 3. Providing both 'start_block_id' and 'start_block_height' simultaneously.
-func (s *BlockHeadersProviderSuite) TestBlockHeadersDataProvider_InvalidArguments() {
-	ctx := context.Background()
-	send := make(chan interface{})
-
-	topic := BlockHeadersTopic
-
-	for _, test := range s.invalidArgumentsTestCases() {
-		s.Run(test.name, func() {
-			provider, err := NewBlockHeadersDataProvider(ctx, s.log, s.api, "dummy-id", topic, test.arguments, send)
-			s.Require().Nil(provider)
-			s.Require().Error(err)
-			s.Require().Contains(err.Error(), test.expectedErrorMsg)
-		})
-	}
+// TestBlockHeadersDataProvider_HappyPath tests the behavior of the block headers data provider
+// when it is configured correctly and operating under normal conditions. It
+// validates that block headers are correctly streamed to the channel and ensures
+// no unexpected errors occur.
+func (s *BlockHeadersProviderSuite) TestBlockHeadersDataProvider_HappyPath() {
+	testHappyPath(
+		s.T(),
+		BlockHeadersTopic,
+		s.factory,
+		s.validBlockHeadersArgumentsTestCases(),
+		func(dataChan chan interface{}) {
+			for _, block := range s.blocks {
+				dataChan <- block.Header
+			}
+		},
+		s.requireBlockHeader,
+	)
 }
 
 // validBlockHeadersArgumentsTestCases defines test happy cases for block headers data providers.
@@ -60,7 +55,10 @@ func (s *BlockHeadersProviderSuite) validBlockHeadersArgumentsTestCases() []test
 		var header commonmodels.BlockHeader
 		header.Build(b.Header)
 
-		expectedResponses[i] = &models.BlockHeaderMessageResponse{Header: &header}
+		expectedResponses[i] = &models.BaseDataProvidersResponse{
+			Topic:   BlockHeadersTopic,
+			Payload: &header,
+		}
 	}
 
 	return []testType{
@@ -113,32 +111,34 @@ func (s *BlockHeadersProviderSuite) validBlockHeadersArgumentsTestCases() []test
 	}
 }
 
-// TestBlockHeadersDataProvider_HappyPath tests the behavior of the block headers data provider
-// when it is configured correctly and operating under normal conditions. It
-// validates that block headers are correctly streamed to the channel and ensures
-// no unexpected errors occur.
-func (s *BlockHeadersProviderSuite) TestBlockHeadersDataProvider_HappyPath() {
-	testHappyPath(
-		s.T(),
-		BlockHeadersTopic,
-		s.factory,
-		s.validBlockHeadersArgumentsTestCases(),
-		func(dataChan chan interface{}) {
-			for _, block := range s.blocks {
-				dataChan <- block.Header
-			}
-		},
-		s.requireBlockHeader,
-	)
-}
-
 // requireBlockHeaders ensures that the received block header information matches the expected data.
 func (s *BlockHeadersProviderSuite) requireBlockHeader(actual interface{}, expected interface{}) {
-	actualResponse, ok := actual.(*models.BlockHeaderMessageResponse)
-	require.True(s.T(), ok, "unexpected response type: %T", actual)
+	expectedResponse, expectedResponsePayload := extractPayload[*commonmodels.BlockHeader](s.T(), expected)
+	actualResponse, actualResponsePayload := extractPayload[*commonmodels.BlockHeader](s.T(), actual)
 
-	expectedResponse, ok := expected.(*models.BlockHeaderMessageResponse)
-	require.True(s.T(), ok, "unexpected response type: %T", expected)
+	s.Require().Equal(expectedResponse.Topic, actualResponse.Topic)
+	s.Require().Equal(expectedResponsePayload, actualResponsePayload)
+}
 
-	s.Require().Equal(expectedResponse.Header, actualResponse.Header)
+// TestBlockHeadersDataProvider_InvalidArguments tests the behavior of the block headers data provider
+// when invalid arguments are provided. It verifies that appropriate errors are returned
+// for missing or conflicting arguments.
+// This test covers the test cases:
+// 1. Missing 'block_status' argument.
+// 2. Invalid 'block_status' argument.
+// 3. Providing both 'start_block_id' and 'start_block_height' simultaneously.
+func (s *BlockHeadersProviderSuite) TestBlockHeadersDataProvider_InvalidArguments() {
+	ctx := context.Background()
+	send := make(chan interface{})
+
+	topic := BlockHeadersTopic
+
+	for _, test := range s.invalidArgumentsTestCases() {
+		s.Run(test.name, func() {
+			provider, err := NewBlockHeadersDataProvider(ctx, s.log, s.api, "dummy-id", topic, test.arguments, send)
+			s.Require().Nil(provider)
+			s.Require().Error(err)
+			s.Require().Contains(err.Error(), test.expectedErrorMsg)
+		})
+	}
 }

engine/access/rest/websockets/data_providers/blocks_provider.go

@@ -86,9 +86,10 @@ func (p *BlocksDataProvider) Run() error {
 				return nil, fmt.Errorf("failed to build block response :%w", err)
 			}
 
-			return &models.BlockMessageResponse{
-				Block: &block,
-			}, nil
+			var response models.BaseDataProvidersResponse
+			response.Build(p.ID(), p.Topic(), &block)
+
+			return &response, nil
 		}),
 	)
 }