YAML parsing test

Author: bolekk

core/scripts/go.sum

@@ -26,8 +26,8 @@ type ProposeGatewayJobInput struct {
 	ServiceCentricFormatEnabled bool              `yaml:"serviceCentricFormatEnabled"`
 	DONs                        []DON             `yaml:"dons"`
 	Services                    []GatewayService  `yaml:"services"`
-	GatewayRequestTimeoutSec    int               `yaml:"gatewayRequestTimeoutSec"`
-	AllowedPorts                []int             `yaml:"allowedPorts"`
+	GatewayRequestTimeoutSec    pkg.Int           `yaml:"gatewayRequestTimeoutSec"`
+	AllowedPorts                []pkg.Int         `yaml:"allowedPorts"`
 	AllowedSchemes              []string          `yaml:"allowedSchemes"`
 	AllowedIPsCIDR              []string          `yaml:"allowedIPsCIDR"`
 	AuthGatewayID               string            `yaml:"authGatewayID"`
@@ -37,7 +37,7 @@ type ProposeGatewayJobInput struct {
 
 type DON struct {
 	Name     string   `yaml:"name"`
-	F        int      `yaml:"f"`
+	F        pkg.Int  `yaml:"f"`
 	Handlers []string `yaml:"handlers"`
 }
 
@@ -66,7 +66,7 @@ var ProposeGatewayJob = operations.NewOperation[ProposeGatewayJobInput, ProposeG
 // When ServiceCentricFormatEnabled is true, it derives the set of unique DON names from
 // input.Services; otherwise it uses the don-centric input.DONs list.
 func proposeGatewayJob(b operations.Bundle, deps ProposeGatewayJobDeps, input ProposeGatewayJobInput) (ProposeGatewayJobOutput, error) {
-	requestTimeoutSec := input.GatewayRequestTimeoutSec
+	requestTimeoutSec := int(input.GatewayRequestTimeoutSec)
 	if requestTimeoutSec == 0 {
 		requestTimeoutSec = defaultGatewayRequestTimeoutSec
 	}
@@ -181,7 +181,7 @@ func buildServiceCentricJob(deps ProposeGatewayJobDeps, input ProposeGatewayJobI
 		DONs:                        dons,
 		Services:                    services,
 		RequestTimeoutSec:           requestTimeoutSec,
-		AllowedPorts:                input.AllowedPorts,
+		AllowedPorts:                toIntSlice(input.AllowedPorts),
 		AllowedSchemes:              input.AllowedSchemes,
 		AllowedIPsCIDR:              input.AllowedIPsCIDR,
 		AuthGatewayID:               input.AuthGatewayID,
@@ -197,7 +197,7 @@ func buildLegacyFormatJob(deps ProposeGatewayJobDeps, input ProposeGatewayJobInp
 		}
 		targetDONs = append(targetDONs, pkg.TargetDON{
 			ID:       ad.Name,
-			F:        ad.F,
+			F:        int(ad.F),
 			Members:  members,
 			Handlers: ad.Handlers,
 		})
@@ -207,7 +207,7 @@ func buildLegacyFormatJob(deps ProposeGatewayJobDeps, input ProposeGatewayJobInp
 		JobName:           "CRE Gateway",
 		TargetDONs:        targetDONs,
 		RequestTimeoutSec: requestTimeoutSec,
-		AllowedPorts:      input.AllowedPorts,
+		AllowedPorts:      toIntSlice(input.AllowedPorts),
 		AllowedSchemes:    input.AllowedSchemes,
 		AllowedIPsCIDR:    input.AllowedIPsCIDR,
 		AuthGatewayID:     input.AuthGatewayID,
@@ -288,6 +288,14 @@ func resolveDONMembers(deps ProposeGatewayJobDeps, input ProposeGatewayJobInput,
 	return members, f, nil
 }
 
+func toIntSlice(vs []pkg.Int) []int {
+	out := make([]int, len(vs))
+	for i, v := range vs {
+		out[i] = int(v)
+	}
+	return out
+}
+
 func parseSelector(sel uint64) (nodev1.ChainType, string, error) {
 	fam, err := chainsel.GetSelectorFamily(sel)
 	if err != nil {

deployment/cre/jobs/operations/propose_gateway_job.go

@@ -425,7 +425,8 @@ type nodeRateLimiterConfig struct {
 }
 
 type httpCapabilitiesHandlerConfig struct {
-	CleanUpPeriodMs int `toml:"CleanUpPeriodMs"`
+	CleanUpPeriodMs int                   `toml:"CleanUpPeriodMs"`
+	NodeRateLimiter nodeRateLimiterConfig `toml:"NodeRateLimiter"`
 }
 
 func newDefaultHTTPCapabilitiesHandler() handler {
@@ -434,6 +435,12 @@ func newDefaultHTTPCapabilitiesHandler() handler {
 		ServiceName: "workflows",
 		Config: httpCapabilitiesHandlerConfig{
 			CleanUpPeriodMs: 10 * 60 * 1000, // 10 minutes
+			NodeRateLimiter: nodeRateLimiterConfig{
+				GlobalBurst:    100,
+				GlobalRPS:      500,
+				PerSenderBurst: 100,
+				PerSenderRPS:   100,
+			},
 		},
 	}
 }

deployment/cre/jobs/pkg/gateway_job.go

@@ -290,6 +290,12 @@ ServiceName = 'workflows'
 [gatewayConfig.Services.Handlers.Config]
 CleanUpPeriodMs = 600000
 
+[gatewayConfig.Services.Handlers.Config.NodeRateLimiter]
+globalBurst = 100
+globalRPS = 500
+perSenderBurst = 100
+perSenderRPS = 100
+
 [[gatewayConfig.Services]]
 ServiceName = 'vault'
 DONs = ['workflow_2']

deployment/cre/jobs/pkg/gateway_job_test.go

@@ -2,6 +2,7 @@ package pkg
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	cldf "github.com/smartcontractkit/chainlink-deployments-framework/deployment"
@@ -25,6 +26,9 @@ type FetchNodeChainConfigsResponse struct {
 }
 
 func FetchNodeChainConfigsFromJD(ctx context.Context, e cldf.Environment, req FetchNodesRequest) ([]FetchNodeChainConfigsResponse, error) {
+	if e.Offchain == nil {
+		return nil, errors.New("offchain client (JD) is not initialized; ensure JD_GRPC or OFFCHAIN_JD_ENDPOINTS_GRPC is set")
+	}
 	resp, err := e.Offchain.ListNodes(ctx, &nodev1.ListNodesRequest{Filter: req.Filters})
 	if err != nil {
 		return nil, fmt.Errorf("failed to list nodes: %w", err)

deployment/cre/jobs/pkg/nodes.go

@@ -29,6 +29,23 @@ type OracleFactoryConfig struct {
 	Network            string   `toml:"network"`              // e.g., "evm"
 }
 
+// Int wraps int so that YAML fields can be populated from either a numeric
+// literal or a quoted string (e.g. after environment-variable substitution).
+type Int int
+
+func (i *Int) UnmarshalYAML(node *yaml.Node) error {
+	v, err := strconv.Atoi(node.Value)
+	if err != nil {
+		return err
+	}
+	*i = Int(v)
+	return nil
+}
+
+func (i Int) MarshalYAML() ([]byte, error) {
+	return []byte(strconv.Itoa(int(i))), nil
+}
+
 type ChainSelector uint64
 
 func (cs *ChainSelector) UnmarshalText(data []byte) error {

deployment/cre/jobs/pkg/types.go

@@ -11,12 +11,14 @@ import (
 	"github.com/pelletier/go-toml/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v3"
 
 	chainsel "github.com/smartcontractkit/chain-selectors"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings"
 	"github.com/smartcontractkit/chainlink-deployments-framework/datastore"
 	cldf "github.com/smartcontractkit/chainlink-deployments-framework/deployment"
+	cldpipelineinput "github.com/smartcontractkit/chainlink-deployments-framework/engine/cld/pipeline/input"
 	"github.com/smartcontractkit/chainlink-protos/job-distributor/v1/job"
 	"github.com/smartcontractkit/chainlink-protos/job-distributor/v1/node"
 	"github.com/smartcontractkit/chainlink/deployment/cre/jobs"
@@ -1753,3 +1755,132 @@ CallLimit = 1_000`, "invalid inputs for CRE settings job spec: invalid wf abcd:
 		})
 	}
 }
+
+func TestProposeJobSpec_GatewayJobYAMLConversion(t *testing.T) {
+	t.Parallel()
+
+	t.Run("service-centric format", func(t *testing.T) {
+		t.Parallel()
+
+		yamlSpec := `
+environment: staging
+domain: cre
+changesets:
+  - job_propose_arbitrary:
+      payload:
+        donName: gateway-don
+        donFilters:
+          - key: don_name
+            value: gateway-don
+          - key: environment
+            value: staging
+          - key: product
+            value: cre
+        jobName: test-gateway-job-svc
+        template: gateway
+        inputs:
+          serviceCentricFormatEnabled: true
+          dons:
+            - name: workflow-don
+              f: 1
+          services:
+            - servicename: workflows
+              handlers:
+                - web-api-capabilities
+                - http-capabilities
+              dons:
+                - workflow-don
+          gatewayRequestTimeoutSec: 10
+          allowedSchemes:
+            - https
+          allowedIPsCIDR:
+            - 10.0.0.0/8
+`
+		var root yaml.Node
+		err := yaml.Unmarshal([]byte(yamlSpec), &root)
+		require.NoError(t, err)
+
+		rootMap, ok := cldpipelineinput.YamlNodeToAny(&root).(map[string]any)
+		require.True(t, ok)
+
+		environment, _ := rootMap["environment"].(string)
+		domain, _ := rootMap["domain"].(string)
+
+		changesetData, err := cldpipelineinput.FindChangesetInData(rootMap["changesets"], "job_propose_arbitrary", "test")
+		require.NoError(t, err)
+
+		changesetMap, ok := changesetData.(map[string]any)
+		require.True(t, ok)
+
+		payload, ok := changesetMap["payload"]
+		require.True(t, ok)
+
+		payloadBytes, err := yaml.Marshal(payload)
+		require.NoError(t, err)
+
+		var parsed jobs.ProposeJobSpecInput
+		err = yaml.Unmarshal(payloadBytes, &parsed)
+		require.NoError(t, err)
+
+		parsed.Environment = environment
+		parsed.Domain = domain
+
+		assert.Equal(t, "staging", parsed.Environment)
+		assert.Equal(t, "cre", parsed.Domain)
+		assert.Equal(t, job_types.Gateway, parsed.Template)
+
+		var gatewayInput operations.ProposeGatewayJobInput
+		err = parsed.Inputs.UnmarshalTo(&gatewayInput)
+		require.NoError(t, err)
+
+		assert.True(t, gatewayInput.ServiceCentricFormatEnabled)
+		require.Len(t, gatewayInput.DONs, 1)
+		assert.Equal(t, "workflow-don", gatewayInput.DONs[0].Name)
+		assert.Equal(t, pkg.Int(1), gatewayInput.DONs[0].F)
+		require.Len(t, gatewayInput.Services, 1)
+		assert.Equal(t, "workflows", gatewayInput.Services[0].ServiceName)
+		assert.Equal(t, []string{"web-api-capabilities", "http-capabilities"}, gatewayInput.Services[0].Handlers)
+		assert.Equal(t, []string{"workflow-don"}, gatewayInput.Services[0].DONs)
+		assert.Equal(t, pkg.Int(10), gatewayInput.GatewayRequestTimeoutSec)
+		assert.Equal(t, []string{"https"}, gatewayInput.AllowedSchemes)
+		assert.Equal(t, []string{"10.0.0.0/8"}, gatewayInput.AllowedIPsCIDR)
+
+		// Build GatewayJob manually; in production member addresses are resolved via JD.
+		gj := pkg.GatewayJob{
+			ServiceCentricFormatEnabled: true,
+			JobName:                     "CRE Gateway",
+			DONs: []pkg.TargetDON{
+				{
+					ID: gatewayInput.DONs[0].Name,
+					F:  int(gatewayInput.DONs[0].F),
+					Members: []pkg.TargetDONMember{
+						{Address: "0xdef456", Name: "mock-node-1 (DON workflow-don)"},
+					},
+				},
+			},
+			Services: []pkg.GatewayServiceConfig{
+				{
+					ServiceName: gatewayInput.Services[0].ServiceName,
+					Handlers:    gatewayInput.Services[0].Handlers,
+					DONs:        gatewayInput.Services[0].DONs,
+				},
+			},
+			RequestTimeoutSec: int(gatewayInput.GatewayRequestTimeoutSec),
+			AllowedSchemes:    gatewayInput.AllowedSchemes,
+			AllowedIPsCIDR:    gatewayInput.AllowedIPsCIDR,
+		}
+
+		require.NoError(t, gj.Validate())
+		assert.True(t, gj.ServiceCentricFormatEnabled)
+		assert.Equal(t, "CRE Gateway", gj.JobName)
+		assert.Equal(t, 10, gj.RequestTimeoutSec)
+		assert.Equal(t, []string{"https"}, gj.AllowedSchemes)
+		assert.Equal(t, []string{"10.0.0.0/8"}, gj.AllowedIPsCIDR)
+		require.Len(t, gj.DONs, 1)
+		assert.Equal(t, "workflow-don", gj.DONs[0].ID)
+		require.Len(t, gj.Services, 1)
+		assert.Equal(t, "workflows", gj.Services[0].ServiceName)
+		assert.Equal(t, []string{"web-api-capabilities", "http-capabilities"}, gj.Services[0].Handlers)
+		assert.Equal(t, []string{"workflow-don"}, gj.Services[0].DONs)
+	})
+}

deployment/cre/jobs/propose_job_spec_test.go

@@ -2,6 +2,7 @@ package offchain
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"slices"
 	"sort"
@@ -32,6 +33,9 @@ const (
 )
 
 func FetchNodesFromJD(ctx context.Context, jd cldf_offchain.Client, filter *nodeapiv1.ListNodesRequest_Filter) (nodes []*nodeapiv1.Node, err error) {
+	if jd == nil {
+		return nil, errors.New("offchain client (JD) is not initialized; ensure JD_GRPC or OFFCHAIN_JD_ENDPOINTS_GRPC is set")
+	}
 	resp, err := jd.ListNodes(ctx, &nodeapiv1.ListNodesRequest{Filter: filter})
 	if err != nil {
 		return nil, fmt.Errorf("failed to list nodes: %w", err)

deployment/cre/pkg/offchain/nodes.go

@@ -454,6 +454,7 @@ require (
 	github.com/streamingfast/logging v0.0.0-20230608130331-f22c91403091 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/supranational/blst v0.3.16 // indirect
+	github.com/suzuki-shunsuke/go-convmap v0.2.1 // indirect
 	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d // indirect
 	github.com/tendermint/go-amino v0.16.0 // indirect
 	github.com/theodesp/go-heaps v0.0.0-20190520121037-88e35354fe0a // indirect