Merge pull request #6907 from The-K-R-O-K/UlianaAndrukhiv/6641-websockets-integration-tests

[Access] Implement integration test for new websockets

Author: peterargue

cmd/access/node_builder/access_node_builder.go

@@ -1458,11 +1458,25 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			defaultConfig.registerDBPruneThreshold,
 			fmt.Sprintf("specifies the number of blocks below the latest stored block height to keep in register db. default: %d", defaultConfig.registerDBPruneThreshold))
 
-		flags.DurationVar(&builder.rpcConf.WebSocketConfig.InactivityTimeout,
+		// websockets config
+		flags.DurationVar(
+			&builder.rpcConf.WebSocketConfig.InactivityTimeout,
 			"websocket-inactivity-timeout",
 			defaultConfig.rpcConf.WebSocketConfig.InactivityTimeout,
-			"specifies the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
-
+			"the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed",
+		)
+		flags.Uint64Var(
+			&builder.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"websocket-max-subscriptions-per-connection",
+			defaultConfig.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"the maximum number of active WebSocket subscriptions allowed per connection",
+		)
+		flags.Float64Var(
+			&builder.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"websocket-max-responses-per-second",
+			defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			fmt.Sprintf("the maximum number of responses that can be sent to a single client per second. Default: %f. if set to 0, no limit is applied to the number of responses per second.", defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond),
+		)
 		flags.BoolVar(
 			&builder.rpcConf.EnableWebSocketsStreamAPI,
 			"experimental-enable-websockets-stream-api",

cmd/observer/node_builder/observer_builder.go

@@ -820,11 +820,25 @@ func (builder *ObserverServiceBuilder) extraFlags() {
 			defaultConfig.registerDBPruneThreshold,
 			fmt.Sprintf("specifies the number of blocks below the latest stored block height to keep in register db. default: %d", defaultConfig.registerDBPruneThreshold))
 
-		flags.DurationVar(&builder.rpcConf.WebSocketConfig.InactivityTimeout,
+		// websockets config
+		flags.DurationVar(
+			&builder.rpcConf.WebSocketConfig.InactivityTimeout,
 			"websocket-inactivity-timeout",
 			defaultConfig.rpcConf.WebSocketConfig.InactivityTimeout,
-			"specifies the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed")
-
+			"the duration a WebSocket connection can remain open without any active subscriptions before being automatically closed",
+		)
+		flags.Uint64Var(
+			&builder.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"websocket-max-subscriptions-per-connection",
+			defaultConfig.rpcConf.WebSocketConfig.MaxSubscriptionsPerConnection,
+			"the maximum number of active WebSocket subscriptions allowed per connection",
+		)
+		flags.Float64Var(
+			&builder.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			"websocket-max-responses-per-second",
+			defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond,
+			fmt.Sprintf("the maximum number of responses that can be sent to a single client per second. Default: %f. if set to 0, no limit is applied to the number of responses per second.", defaultConfig.rpcConf.WebSocketConfig.MaxResponsesPerSecond),
+		)
 		flags.BoolVar(
 			&builder.rpcConf.EnableWebSocketsStreamAPI,
 			"experimental-enable-websockets-stream-api",

engine/access/rest/http/request/address.go engine/access/rest/common/parser/address.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/address_test.go engine/access/rest/common/parser/address_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"strings"

engine/access/rest/http/request/arguments.go engine/access/rest/common/parser/arguments.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"
@@ -7,6 +7,7 @@ import (
 )
 
 const maxArgumentsLength = 100
+const MaxAllowedScriptArguments = 100
 
 type Arguments [][]byte
 
@@ -25,7 +26,7 @@ func (a *Arguments) Parse(raw []string) error {
 	}
 
 	if len(args) > maxArgumentsLength {
-		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments)
+		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments)
 	}
 
 	*a = args

engine/access/rest/http/request/arguments_test.go engine/access/rest/common/parser/arguments_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"
@@ -21,13 +21,13 @@ func TestArguments_InvalidParse(t *testing.T) {
 		assert.EqualError(t, err, "invalid argument encoding: illegal base64 data at input byte 0", a)
 	}
 
-	tooLong := make([]string, maxAllowedScriptArguments+1)
+	tooLong := make([]string, MaxAllowedScriptArguments+1)
 	for i := range tooLong {
 		tooLong[i] = "dGVzdA=="
 	}
 
 	err := arguments.Parse(tooLong)
-	assert.EqualError(t, err, fmt.Sprintf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments))
+	assert.EqualError(t, err, fmt.Sprintf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments))
 }
 
 func TestArguments_ValidParse(t *testing.T) {

engine/access/rest/http/request/proposal_key.go engine/access/rest/common/parser/proposal_key.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/signature_test.go engine/access/rest/common/parser/signature_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"encoding/hex"

engine/access/rest/http/request/signatures.go engine/access/rest/common/parser/signatures.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"fmt"

engine/access/rest/http/request/transaction.go engine/access/rest/common/parser/transaction.go

@@ -1,24 +1,23 @@
-package request
+package parser
 
 import (
 	"fmt"
 	"io"
 
-	"github.com/onflow/flow-go/engine/access/rest/common/parser"
+	"github.com/onflow/flow-go/engine/access/rest/common"
 	"github.com/onflow/flow-go/engine/access/rest/http/models"
 	"github.com/onflow/flow-go/engine/access/rest/util"
 	"github.com/onflow/flow-go/engine/common/rpc/convert"
 	"github.com/onflow/flow-go/model/flow"
 )
 
 const maxAuthorizers = 100
-const maxAllowedScriptArguments = 100
 
 type Transaction flow.TransactionBody
 
 func (t *Transaction) Parse(raw io.Reader, chain flow.Chain) error {
 	var tx models.TransactionsBody
-	err := parseBody(raw, &tx)
+	err := common.ParseBody(raw, &tx)
 	if err != nil {
 		return err
 	}
@@ -35,8 +34,8 @@ func (t *Transaction) Parse(raw io.Reader, chain flow.Chain) error {
 	if len(tx.Authorizers) > maxAuthorizers {
 		return fmt.Errorf("too many authorizers. Maximum authorizers allowed: %d", maxAuthorizers)
 	}
-	if len(tx.Arguments) > maxAllowedScriptArguments {
-		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", maxAllowedScriptArguments)
+	if len(tx.Arguments) > MaxAllowedScriptArguments {
+		return fmt.Errorf("too many arguments. Maximum arguments allowed: %d", MaxAllowedScriptArguments)
 	}
 	if tx.ReferenceBlockId == "" {
 		return fmt.Errorf("reference block not provided")
@@ -90,7 +89,7 @@ func (t *Transaction) Parse(raw io.Reader, chain flow.Chain) error {
 		return fmt.Errorf("invalid transaction script encoding")
 	}
 
-	var blockID parser.ID
+	var blockID ID
 	err = blockID.Parse(tx.ReferenceBlockId)
 	if err != nil {
 		return fmt.Errorf("invalid reference block ID: %w", err)

engine/access/rest/http/request/transaction_test.go engine/access/rest/common/parser/transaction_test.go

@@ -1,4 +1,4 @@
-package request
+package parser
 
 import (
 	"bytes"

engine/access/rest/common/utils.go

@@ -1,5 +1,13 @@
 package common
 
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+)
+
 // SliceToMap converts a slice of strings into a map where each string
 // in the slice becomes a key in the map with the value set to true.
 func SliceToMap(values []string) map[string]bool {
@@ -9,3 +17,69 @@ func SliceToMap(values []string) map[string]bool {
 	}
 	return valueMap
 }
+
+// ParseBody parses the input data into the destination interface and returns any decoding errors
+// updated to be more user-friendly. It also checks that there is exactly one json object in the input
+func ParseBody(raw io.Reader, dst interface{}) error {
+	dec := json.NewDecoder(raw)
+	dec.DisallowUnknownFields()
+
+	err := dec.Decode(&dst)
+	if err != nil {
+		var syntaxError *json.SyntaxError
+		var unmarshalTypeError *json.UnmarshalTypeError
+
+		switch {
+		case errors.As(err, &syntaxError):
+			return fmt.Errorf("request body contains badly-formed JSON (at position %d)", syntaxError.Offset)
+		case errors.Is(err, io.ErrUnexpectedEOF):
+			return fmt.Errorf("request body contains badly-formed JSON")
+		case errors.As(err, &unmarshalTypeError):
+			return fmt.Errorf("request body contains an invalid value for the %q field (at position %d)", unmarshalTypeError.Field, unmarshalTypeError.Offset)
+		case strings.HasPrefix(err.Error(), "json: unknown field "):
+			fieldName := strings.TrimPrefix(err.Error(), "json: unknown field ")
+			return fmt.Errorf("request body contains unknown field %s", fieldName)
+		case errors.Is(err, io.EOF):
+			return fmt.Errorf("request body must not be empty")
+		default:
+			return err
+		}
+	}
+
+	if dst == nil {
+		return fmt.Errorf("request body must not be empty")
+	}
+
+	// verify the request contained exactly one json object
+	err = dec.Decode(&struct{}{})
+	if err != io.EOF {
+		return fmt.Errorf("request body must only contain a single JSON object")
+	}
+
+	return nil
+}
+
+// ParseInterfaceToStrings converts a slice of interface{} to a slice of strings.
+//
+// No errors are expected during normal operations.
+func ParseInterfaceToStrings(value interface{}) ([]string, error) {
+	if strSlice, ok := value.([]string); ok {
+		return strSlice, nil
+	}
+
+	interfaceSlice, ok := value.([]interface{})
+	if !ok {
+		return nil, fmt.Errorf("value must be an array. got %T", value)
+	}
+
+	result := make([]string, len(interfaceSlice))
+	for i, v := range interfaceSlice {
+		str, ok := v.(string)
+		if !ok {
+			return nil, fmt.Errorf("value must be an array of strings. got %T", v)
+		}
+		result[i] = str
+	}
+
+	return result, nil
+}

engine/access/rest/common/utils_test.go

@@ -0,0 +1,47 @@
+package common
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_ParseBody(t *testing.T) {
+
+	invalid := []struct {
+		in  string
+		err string
+	}{
+		{"{f}", "request body contains badly-formed JSON (at position 2)"},
+		{"foo", "request body contains badly-formed JSON (at position 2)"},
+		{"", "request body must not be empty"},
+		{`{"foo": "bar"`, "request body contains badly-formed JSON"},
+		{`{"foo": "bar" "foo2":"bar2"}`, "request body contains badly-formed JSON (at position 15)"},
+		{`{"foo":"bar"}, {}`, "request body must only contain a single JSON object"},
+		{`[][]`, "request body must only contain a single JSON object"},
+	}
+
+	for i, test := range invalid {
+		readerIn := strings.NewReader(test.in)
+		var out interface{}
+		err := ParseBody(readerIn, out)
+		assert.EqualError(t, err, test.err, fmt.Sprintf("test #%d failed", i))
+	}
+
+	type body struct {
+		Foo string
+		Bar bool
+		Zoo uint64
+	}
+	var b body
+	err := ParseBody(strings.NewReader(`{ "foo": "test", "bar": true }`), &b)
+	assert.NoError(t, err)
+	assert.Equal(t, b.Bar, true)
+	assert.Equal(t, b.Foo, "test")
+	assert.Equal(t, b.Zoo, uint64(0))
+
+	err = ParseBody(strings.NewReader(`{ "foo": false }`), &b)
+	assert.EqualError(t, err, `request body contains an invalid value for the "Foo" field (at position 14)`)
+}

engine/access/rest/http/request/create_transaction.go

@@ -4,6 +4,7 @@ import (
 	"io"
 
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/model/flow"
 )
 
@@ -22,7 +23,7 @@ func (c *CreateTransaction) Build(r *common.Request) error {
 }
 
 func (c *CreateTransaction) Parse(rawTransaction io.Reader, chain flow.Chain) error {
-	var tx Transaction
+	var tx parser.Transaction
 	err := tx.Parse(rawTransaction, chain)
 	if err != nil {
 		return err

engine/access/rest/http/request/get_account.go

@@ -2,6 +2,7 @@ package request
 
 import (
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/model/flow"
 )
 
@@ -32,7 +33,7 @@ func (g *GetAccount) Build(r *common.Request) error {
 }
 
 func (g *GetAccount) Parse(rawAddress string, rawHeight string, chain flow.Chain) error {
-	address, err := ParseAddress(rawAddress, chain)
+	address, err := parser.ParseAddress(rawAddress, chain)
 	if err != nil {
 		return err
 	}

engine/access/rest/http/request/get_account_balance.go

@@ -2,6 +2,7 @@ package request
 
 import (
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/model/flow"
 )
 
@@ -33,7 +34,7 @@ func (g *GetAccountBalance) Parse(
 	rawHeight string,
 	chain flow.Chain,
 ) error {
-	address, err := ParseAddress(rawAddress, chain)
+	address, err := parser.ParseAddress(rawAddress, chain)
 	if err != nil {
 		return err
 	}

engine/access/rest/http/request/get_account_key.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/onflow/flow-go/engine/access/rest/common"
+	"github.com/onflow/flow-go/engine/access/rest/common/parser"
 	"github.com/onflow/flow-go/engine/access/rest/util"
 	"github.com/onflow/flow-go/model/flow"
 )
@@ -41,7 +42,7 @@ func (g *GetAccountKey) Parse(
 	rawHeight string,
 	chain flow.Chain,
 ) error {
-	address, err := ParseAddress(rawAddress, chain)
+	address, err := parser.ParseAddress(rawAddress, chain)
 	if err != nil {
 		return err
 	}