Merge branch 'master' into subscribe-block-digest-endpoints

access/grpc/grpc_test.go

@@ -2382,6 +2382,160 @@ func assertNoBlocks[T any](t *testing.T, blocksCh <-chan T, done func()) {
 	}
 }
 
+func TestClient_SendAndSubscribeTransactionStatuses(t *testing.T) {
+	transactions := test.TransactionGenerator()
+
+	generateTransactionStatusResponses := func(count uint64, encodingVersion flow.EventEncodingVersion) []*access.SendAndSubscribeTransactionStatusesResponse {
+		var resTransactionResults []*access.SendAndSubscribeTransactionStatusesResponse
+		results := test.TransactionResultGenerator(encodingVersion)
+
+		for i := uint64(0); i < count; i++ {
+			expectedResult := results.New()
+			transactionResult, _ := convert.TransactionResultToMessage(expectedResult, encodingVersion)
+
+			response := &access.SendAndSubscribeTransactionStatusesResponse{
+				TransactionResults: transactionResult,
+				MessageIndex:       i,
+			}
+
+			resTransactionResults = append(resTransactionResults, response)
+		}
+
+		return resTransactionResults
+	}
+
+	t.Run("Happy Path - CCF", clientTest(func(t *testing.T, ctx context.Context, rpc *mocks.MockRPCClient, c *BaseClient) {
+		responseCount := uint64(100)
+		tx := transactions.New()
+
+		ctx, cancel := context.WithCancel(ctx)
+		stream := &mockTransactionStatusesClientStream{
+			ctx:       ctx,
+			responses: generateTransactionStatusResponses(responseCount, flow.EventEncodingVersionCCF),
+		}
+
+		rpc.On("SendAndSubscribeTransactionStatuses", ctx, mock.Anything).Return(stream, nil)
+
+		txResultCh, errCh, err := c.SendAndSubscribeTransactionStatuses(ctx, *tx)
+		require.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go assertNoErrors(t, errCh, wg.Done)
+
+		expectedCounter := uint64(0)
+
+		for i := uint64(0); i < responseCount; i++ {
+			actualTxResult := <-txResultCh
+			expectedTxResult, err := convert.MessageToTransactionResult(stream.responses[i].GetTransactionResults(), DefaultClientOptions().jsonOptions)
+			require.NoError(t, err)
+			require.Equal(t, expectedTxResult, actualTxResult)
+			require.Equal(t, expectedCounter, stream.responses[i].MessageIndex)
+
+			expectedCounter++
+		}
+		cancel()
+
+		wg.Wait()
+	}))
+
+	t.Run("Happy Path - JSON-CDC", clientTest(func(t *testing.T, ctx context.Context, rpc *mocks.MockRPCClient, c *BaseClient) {
+		responseCount := uint64(100)
+		tx := transactions.New()
+
+		ctx, cancel := context.WithCancel(ctx)
+		stream := &mockTransactionStatusesClientStream{
+			ctx:       ctx,
+			responses: generateTransactionStatusResponses(responseCount, flow.EventEncodingVersionJSONCDC),
+		}
+
+		rpc.On("SendAndSubscribeTransactionStatuses", ctx, mock.Anything).Return(stream, nil)
+
+		txResultCh, errCh, err := c.SendAndSubscribeTransactionStatuses(ctx, *tx)
+		require.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go assertNoErrors(t, errCh, wg.Done)
+
+		expectedCounter := uint64(0)
+		for i := uint64(0); i < responseCount; i++ {
+			actualTxResult := <-txResultCh
+			expectedTxResult, err := convert.MessageToTransactionResult(stream.responses[i].GetTransactionResults(), DefaultClientOptions().jsonOptions)
+			require.NoError(t, err)
+			require.Equal(t, expectedTxResult, actualTxResult)
+			require.Equal(t, expectedCounter, stream.responses[i].MessageIndex)
+
+			expectedCounter++
+		}
+		cancel()
+
+		wg.Wait()
+	}))
+
+	t.Run("Stream returns error", clientTest(func(t *testing.T, ctx context.Context, rpc *mocks.MockRPCClient, c *BaseClient) {
+		ctx, cancel := context.WithCancel(ctx)
+		stream := &mockTransactionStatusesClientStream{
+			ctx: ctx,
+			err: status.Error(codes.Internal, "internal error"),
+		}
+
+		rpc.
+			On("SendAndSubscribeTransactionStatuses", ctx, mock.Anything).
+			Return(stream, nil)
+
+		txResultChan, errCh, err := c.SendAndSubscribeTransactionStatuses(ctx, flow.Transaction{})
+		require.NoError(t, err)
+
+		wg := sync.WaitGroup{}
+		wg.Add(1)
+		go assertNoTxResults(t, txResultChan, wg.Done)
+
+		errorCount := 0
+		for e := range errCh {
+			require.Error(t, e)
+			require.ErrorIs(t, e, stream.err)
+			errorCount += 1
+		}
+		cancel()
+
+		require.Equalf(t, 1, errorCount, "only 1 error is expected")
+
+		wg.Wait()
+	}))
+
+}
+
+type mockTransactionStatusesClientStream struct {
+	grpc.ClientStream
+
+	ctx       context.Context
+	err       error
+	offset    int
+	responses []*access.SendAndSubscribeTransactionStatusesResponse
+}
+
+func (m *mockTransactionStatusesClientStream) Recv() (*access.SendAndSubscribeTransactionStatusesResponse, error) {
+	if m.err != nil {
+		return nil, m.err
+	}
+
+	if m.offset >= len(m.responses) {
+		<-m.ctx.Done()
+		return nil, io.EOF
+	}
+	defer func() { m.offset++ }()
+
+	return m.responses[m.offset], nil
+}
+
+func assertNoTxResults[TxStatus any](t *testing.T, txResultChan <-chan TxStatus, done func()) {
+	defer done()
+	for range txResultChan {
+		require.FailNow(t, "should not receive txStatus")
+	}
+}
+
 func TestClient_SubscribeBlockDigest(t *testing.T) {
 	blockHeaders := test.BlockHeaderGenerator()