Ccip-3535 env abstraction updated

integration-tests/deployment/ccip/add_chain_test.go

@@ -24,8 +24,7 @@ import (
 
 func TestAddChainInbound(t *testing.T) {
 	// 4 chains where the 4th is added after initial deployment.
-	e := NewEnvironmentWithCRAndJobs(t, logger.TestLogger(t), 4)
-	require.Equal(t, len(e.Nodes), 5)
+	e := NewMemoryEnvironmentWithJobs(t, logger.TestLogger(t), 4)
 	state, err := LoadOnchainState(e.Env, e.Ab)
 	require.NoError(t, err)
 	// Take first non-home chain as the new chain.
@@ -127,6 +126,9 @@ func TestAddChainInbound(t *testing.T) {
 		ExecuteProposal(t, e.Env, chainInboundExec, state, sel)
 	}
 
+	replayBlocks, err := LatestBlocksByChain(testcontext.Get(t), e.Env.Chains)
+	require.NoError(t, err)
+
 	// Now configure the new chain using deployer key (not transferred to timelock yet).
 	var offRampEnables []offramp.OffRampSourceChainConfigArgs
 	for _, source := range initialDeploy {
@@ -167,7 +169,7 @@ func TestAddChainInbound(t *testing.T) {
 	}
 	// Ensure job related logs are up to date.
 	time.Sleep(30 * time.Second)
-	require.NoError(t, ReplayAllLogs(e.Nodes, e.Env.Chains))
+	ReplayLogs(t, e.Env.Offchain, replayBlocks)
 
 	// TODO: Send via all inbound lanes and use parallel helper
 	// Now that the proposal has been executed we expect to be able to send traffic to this new 4th chain.

integration-tests/deployment/ccip/add_lane_test.go

@@ -16,7 +16,7 @@ func TestAddLane(t *testing.T) {
 	// TODO: The offchain code doesn't yet support partial lane
 	// enablement, need to address then re-enable this test.
 	t.Skip()
-	e := NewEnvironmentWithCRAndJobs(t, logger.TestLogger(t), 3)
+	e := NewMemoryEnvironmentWithJobs(t, logger.TestLogger(t), 3)
 	// Here we have CR + nodes set up, but no CCIP contracts deployed.
 	state, err := LoadOnchainState(e.Env, e.Ab)
 	require.NoError(t, err)

integration-tests/deployment/ccip/changeset/2_initial_deploy_test.go

@@ -20,10 +20,8 @@ import (
 func Test0002_InitialDeploy(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	ctx := ccdeploy.Context(t)
-	tenv := ccdeploy.NewEnvironmentWithCRAndFeeds(t, lggr, 3)
+	tenv := ccdeploy.NewMemoryEnvironment(t, lggr, 3)
 	e := tenv.Env
-	nodes := tenv.Nodes
-	chains := e.Chains
 
 	state, err := ccdeploy.LoadOnchainState(tenv.Env, tenv.Ab)
 	require.NoError(t, err)
@@ -52,7 +50,7 @@ func Test0002_InitialDeploy(t *testing.T) {
 	require.NoError(t, err)
 
 	// Ensure capreg logs are up to date.
-	require.NoError(t, ccdeploy.ReplayAllLogs(nodes, chains))
+	ccdeploy.ReplayLogs(t, e.Offchain, tenv.ReplayBlocks)
 
 	// Apply the jobs.
 	for nodeID, jobs := range output.JobSpecs {

integration-tests/deployment/ccip/test_helpers.go

@@ -9,6 +9,7 @@
 	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/smartcontractkit/chainlink-testing-framework/lib/utils/testcontext"
@@ -64,11 +65,43 @@
 	Env          deployment.Environment
 	HomeChainSel uint64
 	FeedChainSel uint64
+	ReplayBlocks map[uint64]uint64
 }
 
-type DeployedTestEnvironment struct {
-	DeployedEnv
-	Nodes map[string]memory.Node
+func (e *DeployedEnv) SetupJobs(t *testing.T) {
+	ctx := testcontext.Get(t)
+	jbs, err := NewCCIPJobSpecs(e.Env.NodeIDs, e.Env.Offchain)
+	require.NoError(t, err)
+	for nodeID, jobs := range jbs {
+		for _, job := range jobs {
+			// Note these auto-accept
+			_, err := e.Env.Offchain.ProposeJob(ctx,
+				&jobv1.ProposeJobRequest{
+					NodeId: nodeID,
+					Spec:   job,
+				})
+			require.NoError(t, err)
+		}
+	}
+	// Wait for plugins to register filters?
+	// TODO: Investigate how to avoid.
+	time.Sleep(30 * time.Second)
+	ReplayLogs(t, e.Env.Offchain, e.ReplayBlocks)
+}
+
+func ReplayLogs(t *testing.T, oc deployment.OffchainClient, replayBlocks map[uint64]uint64) {
+	switch oc.(type) {
+	case *memory.JobClient:
+		jobclient, ok := oc.(*memory.JobClient)
+		require.True(t, ok)
+		require.NoError(t, jobclient.ReplayLogs(replayBlocks))
+	case *devenv.JobDistributor:
+		jobclient, ok := oc.(*devenv.JobDistributor)
+		require.True(t, ok)
+		require.NoError(t, jobclient.ReplayLogs(replayBlocks))
+	default:
+		t.Fatalf("unsupported offchain client type %T", oc)
+	}
 }
 
 func SetUpHomeAndFeedChains(t *testing.T, lggr logger.Logger, homeChainSel, feedChainSel uint64, chains map[uint64]deployment.Chain) (deployment.AddressBook, deployment.CapabilityRegistryConfig) {
@@ -85,11 +118,24 @@
 	}
 }
 
-// NewEnvironmentWithCRAndFeeds creates a new CCIP environment
+func LatestBlocksByChain(ctx context.Context, chains map[uint64]deployment.Chain) (map[uint64]uint64, error) {
+	latestBlocks := make(map[uint64]uint64)
+	for _, chain := range chains {
+		latesthdr, err := chain.Client.HeaderByNumber(ctx, nil)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to get latest header for chain %d", chain.Selector)
+		}
+		block := latesthdr.Number.Uint64()
+		latestBlocks[chain.Selector] = block
+	}
+	return latestBlocks, nil
+}
+
+// NewMemoryEnvironment creates a new CCIP environment
 // with capreg, feeds and nodes set up.
-func NewEnvironmentWithCRAndFeeds(t *testing.T, lggr logger.Logger, numChains int) DeployedTestEnvironment {
+func NewMemoryEnvironment(t *testing.T, lggr logger.Logger, numChains int) DeployedEnv {
 	require.GreaterOrEqual(t, numChains, 2, "numChains must be at least 2 for home and feed chains")
-	ctx := Context(t)
+	ctx := testcontext.Get(t)
 	chains := memory.NewMemoryChains(t, numChains)
 	// Lower chainSel is home chain.
 	var chainSels []uint64
@@ -103,6 +149,8 @@
 	// Take lowest for determinism.
 	homeChainSel := chainSels[HomeChainIndex]
 	feedSel := chainSels[FeedChainIndex]
+	replayBlocks, err := LatestBlocksByChain(ctx, chains)
+	require.NoError(t, err)
 	ab, capReg := SetUpHomeAndFeedChains(t, lggr, homeChainSel, feedSel, chains)
 
 	nodes := memory.NewNodes(t, zapcore.InfoLevel, chains, 4, 1, capReg)
@@ -114,55 +162,21 @@
 	}
 
 	e := memory.NewMemoryEnvironmentFromChainsNodes(t, lggr, chains, nodes)
-	return DeployedTestEnvironment{
-		DeployedEnv: DeployedEnv{
-			Ab:           ab,
-			Env:          e,
-			HomeChainSel: homeChainSel,
-			FeedChainSel: feedSel,
-		},
-		Nodes: nodes,
+	return DeployedEnv{
+		Ab:           ab,
+		Env:          e,
+		HomeChainSel: homeChainSel,
+		FeedChainSel: feedSel,
+		ReplayBlocks: replayBlocks,
 	}
 }
 
-func NewEnvironmentWithCRAndJobs(t *testing.T, lggr logger.Logger, numChains int) DeployedTestEnvironment {
-	ctx := Context(t)
-	e := NewEnvironmentWithCRAndFeeds(t, lggr, numChains)
-	jbs, err := NewCCIPJobSpecs(e.Env.NodeIDs, e.Env.Offchain)
-	require.NoError(t, err)
-	for nodeID, jobs := range jbs {
-		for _, job := range jobs {
-			// Note these auto-accept
-			_, err := e.Env.Offchain.ProposeJob(ctx,
-				&jobv1.ProposeJobRequest{
-					NodeId: nodeID,
-					Spec:   job,
-				})
-			require.NoError(t, err)
-		}
-	}
-	// Wait for plugins to register filters?
-	// TODO: Investigate how to avoid.
-	time.Sleep(30 * time.Second)
-
-	// Ensure job related logs are up to date.
-	require.NoError(t, ReplayAllLogs(e.Nodes, e.Env.Chains))
+func NewMemoryEnvironmentWithJobs(t *testing.T, lggr logger.Logger, numChains int) DeployedEnv {
+	e := NewMemoryEnvironment(t, lggr, numChains)
+	e.SetupJobs(t)
 	return e
 }
 
-func ReplayAllLogs(nodes map[string]memory.Node, chains map[uint64]deployment.Chain) error {
-	blockBySel := make(map[uint64]uint64)
-	for sel := range chains {
-		blockBySel[sel] = 1
-	}
-	for _, node := range nodes {
-		if err := node.ReplayLogs(blockBySel); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func SendRequest(t *testing.T, e deployment.Environment, state CCIPOnChainState, src, dest uint64, testRouter bool) uint64 {
 	msg := router.ClientEVM2AnyMessage{
 		Receiver:     common.LeftPadBytes(state.Chains[dest].Receiver.Address().Bytes(), 32),
@@ -228,8 +242,8 @@
 	return errGrp.Wait()
 }
 
-func NewDeployedLocalDevEnvironment(t *testing.T, lggr logger.Logger) DeployedLocalDevEnvironment {
-	ctx := Context(t)
+func NewLocalDevEnvironment(t *testing.T, lggr logger.Logger) DeployedEnv {
+	ctx := testcontext.Get(t)
 	// create a local docker environment with simulated chains and job-distributor
 	// we cannot create the chainlink nodes yet as we need to deploy the capability registry first
 	envConfig, testEnv, cfg := devenv.CreateDockerEnv(t)
@@ -243,6 +257,8 @@
 	require.NotEmpty(t, homeChainSel, "homeChainSel should not be empty")
 	feedSel := envConfig.FeedChainSelector
 	require.NotEmpty(t, feedSel, "feedSel should not be empty")
+	replayBlocks, err := LatestBlocksByChain(ctx, chains)
+	require.NoError(t, err)
 	ab, capReg := SetUpHomeAndFeedChains(t, lggr, homeChainSel, feedSel, chains)
 
 	// start the chainlink nodes with the CR address
@@ -252,20 +268,16 @@
 	e, don, err := devenv.NewEnvironment(ctx, lggr, *envConfig)
 	require.NoError(t, err)
 	require.NotNil(t, e)
-	require.NotNil(t, don)
 	zeroLogLggr := logging.GetTestLogger(t)
 	// fund the nodes
 	devenv.FundNodes(t, zeroLogLggr, testEnv, cfg, don.PluginNodes())
 
-	return DeployedLocalDevEnvironment{
-		DeployedEnv: DeployedEnv{
-			Ab:           ab,
-			Env:          *e,
-			HomeChainSel: homeChainSel,
-			FeedChainSel: feedSel,
-		},
-		DON:     don,
-		testEnv: testEnv,
+	return DeployedEnv{
+		Ab:           ab,
+		Env:          *e,
+		HomeChainSel: homeChainSel,
+		FeedChainSel: feedSel,
+		ReplayBlocks: replayBlocks,
 	}
 }
 

integration-tests/deployment/devenv/build_env.go

@@ -11,10 +11,13 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	chainselectors "github.com/smartcontractkit/chain-selectors"
+	"github.com/smartcontractkit/chainlink-testing-framework/lib/utils/testcontext"
 	"github.com/stretchr/testify/require"
 	"github.com/subosito/gotenv"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/smartcontractkit/chainlink-testing-framework/lib/utils/conversions"
 	"github.com/smartcontractkit/chainlink-testing-framework/seth"
@@ -207,7 +210,10 @@ func StartChainlinkNodes(
 			InternalIP: n.API.InternalIP(),
 		}
 	}
-	envConfig.nodeInfo = nodeInfo
+	if envConfig == nil {
+		envConfig = &EnvironmentConfig{}
+	}
+	envConfig.JDConfig.nodeInfo = nodeInfo
 	return nil
 }
 
@@ -345,3 +351,28 @@ func CreateChainConfigFromNetworks(
 	}
 	return chains
 }
+
+// RestartChainlinkNodes restarts the chainlink nodes in the test environment
+// ignore lint
+// nolint:usage
+func RestartChainlinkNodes(t *testing.T, env *test_env.CLClusterTestEnv) error {
+	errGrp := errgroup.Group{}
+	if env == nil || env.ClCluster == nil {
+		return errors.Wrap(errors.New("no testenv or clcluster found "), "error restarting node")
+	}
+	for _, n := range env.ClCluster.Nodes {
+		n := n
+		errGrp.Go(func() error {
+			if err := n.Container.Terminate(testcontext.Get(t)); err != nil {
+				return err
+			}
+			err := n.RestartContainer()
+			if err != nil {
+				return err
+			}
+			return nil
+		})
+
+	}
+	return errGrp.Wait()
+}

integration-tests/deployment/devenv/don.go

@@ -160,7 +160,7 @@ type Node struct {
 // It expects bootstrap nodes to have label with key "type" and value as "bootstrap".
 // It fetches the account address, peer id, and OCR2 key bundle id and creates the JobDistributorChainConfig.
 func (n *Node) CreateCCIPOCRSupportedChains(ctx context.Context, chains []ChainConfig, jd JobDistributor) error {
-	for _, chain := range chains {
+	for i, chain := range chains {
 		chainId := strconv.FormatUint(chain.ChainID, 10)
 		accountAddr, err := n.gqlClient.FetchAccountAddress(ctx, chainId)
 		if err != nil {
@@ -198,10 +198,11 @@ func (n *Node) CreateCCIPOCRSupportedChains(ctx context.Context, chains []ChainC
 		}
 		// JD silently fails to update nodeChainConfig. Therefore, we fetch the node config and
 		// if it's not updated , we retry creating the chain config.
-		// as a workaround, we keep trying creating the chain config for 3 times until it's created
+		// as a workaround, we keep trying creating the chain config for 5 times until it's created
 		retryCount := 1
-		for retryCount < 3 {
-			err = n.gqlClient.CreateJobDistributorChainConfig(ctx, client.JobDistributorChainConfigInput{
+		created := false
+		for retryCount < 5 {
+			chainConfigId, err := n.gqlClient.CreateJobDistributorChainConfig(ctx, client.JobDistributorChainConfigInput{
 				JobDistributorID: n.JDId,
 				ChainID:          chainId,
 				ChainType:        chain.ChainType,
@@ -225,22 +226,51 @@ func (n *Node) CreateCCIPOCRSupportedChains(ctx context.Context, chains []ChainC
 			if err != nil {
 				return fmt.Errorf("failed to list node chain configs for node %s: %w", n.Name, err)
 			}
-			if nodeChainConfigs != nil && len(nodeChainConfigs.ChainConfigs) > 0 {
+			if nodeChainConfigs != nil && len(nodeChainConfigs.ChainConfigs) == i+1 {
+				created = true
 				break
 			}
+			// delete the node chain config if it's not updated properly
+			err = n.gqlClient.DeleteJobDistributorChainConfig(ctx, chainConfigId)
+			if err != nil {
+				return fmt.Errorf("failed to delete job distributor chain config for node %s: %w", n.Name, err)
+			}
 			retryCount++
 		}
+		if !created {
+			return fmt.Errorf("failed to create CCIPOCR2SupportedChains for node %s", n.Name)
+		}
 	}
 	return nil
 }
 
-func (n *Node) AcceptJob(ctx context.Context, id string) error {
-	spec, err := n.gqlClient.ApproveJobProposalSpec(ctx, id, false)
+// AcceptJob accepts the job proposal for the given job proposal spec
+func (n *Node) AcceptJob(ctx context.Context, spec string) error {
+	// fetch JD to get the job proposals
+	jd, err := n.gqlClient.GetJobDistributor(ctx, n.JDId)
+	if err != nil {
+		return err
+	}
+	if jd.GetJobProposals() == nil {
+		return fmt.Errorf("no job proposals found for node %s", n.Name)
+	}
+	// locate the job proposal id for the given job spec
+	var idToAccept string
+	for _, jp := range jd.JobProposals {
+		if jp.LatestSpec.Definition == spec {
+			idToAccept = jp.Id
+			break
+		}
+	}
+	if idToAccept == "" {
+		return fmt.Errorf("no job proposal found for job spec %s", spec)
+	}
+	aaprovedSpec, err := n.gqlClient.ApproveJobProposalSpec(ctx, idToAccept, false)
 	if err != nil {
 		return err
 	}
-	if spec == nil {
-		return fmt.Errorf("no job proposal spec found for job id %s", id)
+	if aaprovedSpec == nil {
+		return fmt.Errorf("no job proposal spec found for job id %s", idToAccept)
 	}
 	return nil
 }