Test job creation and replacement (#9934)

* Initial draft

* - Add OCR, OCR2, Automation, VRF and cron tests

* - Update keeper test

* Fix merge

* - Split functions into DeleteJobs and DeleteBridges

* Use standardCleanup

* Fix merge conflict

* Update error style

* Remove unused method

integration-tests/actions/ocr2_helpers_local.go

@@ -19,6 +19,7 @@ import (
 	"gopkg.in/guregu/null.v4"
 
 	"github.com/smartcontractkit/chainlink-testing-framework/docker/test_env"
+
 	"github.com/smartcontractkit/chainlink/integration-tests/client"
 	"github.com/smartcontractkit/chainlink/integration-tests/contracts"
 	"github.com/smartcontractkit/chainlink/v2/core/services/job"
@@ -273,3 +274,49 @@ func GetOracleIdentitiesWithKeyIndexLocal(
 
 	return S, oracleIdentities, eg.Wait()
 }
+
+// DeleteJobs will delete ALL jobs from the nodes
+func DeleteJobs(nodes []*client.ChainlinkClient) error {
+	for _, node := range nodes {
+		if node == nil {
+			return fmt.Errorf("found a nil chainlink node in the list of chainlink nodes while tearing down: %v", nodes)
+		}
+		jobs, _, err := node.ReadJobs()
+		if err != nil {
+			return fmt.Errorf("error reading jobs from chainlink node: %w", err)
+		}
+		for _, maps := range jobs.Data {
+			if _, ok := maps["id"]; !ok {
+				return fmt.Errorf("error reading job id from chainlink node's jobs %+v", jobs.Data)
+			}
+			id := maps["id"].(string)
+			_, err2 := node.DeleteJob(id)
+			if err2 != nil {
+				return fmt.Errorf("error deleting job from chainlink node: %w", err)
+			}
+		}
+	}
+	return nil
+}
+
+// DeleteBridges will delete ALL bridges from the nodes
+func DeleteBridges(nodes []*client.ChainlinkClient) error {
+	for _, node := range nodes {
+		if node == nil {
+			return fmt.Errorf("found a nil chainlink node in the list of chainlink nodes while tearing down: %v", nodes)
+		}
+
+		bridges, _, err := node.ReadBridges()
+		if err != nil {
+			return err
+		}
+		for _, b := range bridges.Data {
+			_, err = node.DeleteBridge(b.Attributes.Name)
+			if err != nil {
+				return err
+			}
+		}
+
+	}
+	return nil
+}

integration-tests/client/chainlink.go

@@ -5,12 +5,11 @@ import (
 	"fmt"
 	"math/big"
 	"net/http"
+	"os"
 	"strings"
 	"sync"
 	"time"
 
-	"os"
-
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/go-resty/resty/v2"
 	"github.com/rs/zerolog"
@@ -303,8 +302,8 @@ func (c *ChainlinkClient) ReadBridge(name string) (*BridgeType, *http.Response,
 }
 
 // ReadBridges reads bridges from the Chainlink node
-func (c *ChainlinkClient) ReadBridges() (*ResponseSlice, *resty.Response, error) {
-	result := &ResponseSlice{}
+func (c *ChainlinkClient) ReadBridges() (*Bridges, *resty.Response, error) {
+	result := &Bridges{}
 	c.l.Info().Str(NodeURL, c.Config.URL).Msg("Getting all bridges")
 	resp, err := c.APIClient.R().
 		SetResult(&result).

integration-tests/client/chainlink_models.go

@@ -107,6 +107,11 @@ type BridgeTypeData struct {
 	Attributes BridgeTypeAttributes `json:"attributes"`
 }
 
+// Bridges is the model that represents the bridges when read on a Chainlink node
+type Bridges struct {
+	Data []BridgeTypeData `json:"data"`
+}
+
 // BridgeTypeAttributes is the model that represents the bridge when read or created on a Chainlink node
 type BridgeTypeAttributes struct {
 	Name        string `json:"name"`

integration-tests/smoke/cron_test.go

@@ -60,3 +60,74 @@ func TestCronBasic(t *testing.T) {
 		}
 	}, "2m", "3s").Should(gomega.Succeed())
 }
+
+func TestCronJobReplacement(t *testing.T) {
+	t.Parallel()
+	l := logging.GetTestLogger(t)
+
+	env, err := test_env.NewCLTestEnvBuilder().
+		WithTestLogger(t).
+		WithGeth().
+		WithMockAdapter().
+		WithCLNodes(1).
+		WithStandardCleanup().
+		Build()
+	require.NoError(t, err)
+
+	err = env.MockAdapter.SetAdapterBasedIntValuePath("/variable", []string{http.MethodGet, http.MethodPost}, 5)
+	require.NoError(t, err, "Setting value path in mockserver shouldn't fail")
+
+	bta := &client.BridgeTypeAttributes{
+		Name:        fmt.Sprintf("variable-%s", uuid.NewString()),
+		URL:         fmt.Sprintf("%s/variable", env.MockAdapter.InternalEndpoint),
+		RequestData: "{}",
+	}
+	err = env.ClCluster.Nodes[0].API.MustCreateBridge(bta)
+	require.NoError(t, err, "Creating bridge in chainlink node shouldn't fail")
+
+	// CRON job creation and replacement
+	job, err := env.ClCluster.Nodes[0].API.MustCreateJob(&client.CronJobSpec{
+		Schedule:          "CRON_TZ=UTC * * * * * *",
+		ObservationSource: client.ObservationSourceSpecBridge(bta),
+	})
+	require.NoError(t, err, "Creating Cron Job in chainlink node shouldn't fail")
+
+	gom := gomega.NewWithT(t)
+	gom.Eventually(func(g gomega.Gomega) {
+		jobRuns, err := env.ClCluster.Nodes[0].API.MustReadRunsByJob(job.Data.ID)
+		if err != nil {
+			l.Info().Err(err).Msg("error while waiting for job runs")
+		}
+		g.Expect(err).ShouldNot(gomega.HaveOccurred(), "Reading Job run data shouldn't fail")
+
+		g.Expect(len(jobRuns.Data)).Should(gomega.BeNumerically(">=", 5), "Expected number of job runs to be greater than 5, but got %d", len(jobRuns.Data))
+
+		for _, jr := range jobRuns.Data {
+			g.Expect(jr.Attributes.Errors).Should(gomega.Equal([]interface{}{nil}), "Job run %s shouldn't have errors", jr.ID)
+		}
+	}, "3m", "3s").Should(gomega.Succeed())
+
+	err = env.ClCluster.Nodes[0].API.MustDeleteJob(job.Data.ID)
+	require.NoError(t, err)
+
+	job, err = env.ClCluster.Nodes[0].API.MustCreateJob(&client.CronJobSpec{
+		Schedule:          "CRON_TZ=UTC * * * * * *",
+		ObservationSource: client.ObservationSourceSpecBridge(bta),
+	})
+	require.NoError(t, err, "Recreating Cron Job in chainlink node shouldn't fail")
+
+	gom.Eventually(func(g gomega.Gomega) {
+		jobRuns, err := env.ClCluster.Nodes[0].API.MustReadRunsByJob(job.Data.ID)
+		if err != nil {
+			l.Info().Err(err).Msg("error while waiting for job runs")
+		}
+		g.Expect(err).ShouldNot(gomega.HaveOccurred(), "Reading Job run data shouldn't fail")
+
+		g.Expect(len(jobRuns.Data)).Should(gomega.BeNumerically(">=", 5), "Expected number of job runs to be greater than 5, but got %d", len(jobRuns.Data))
+
+		for _, jr := range jobRuns.Data {
+			g.Expect(jr.Attributes.Errors).Should(gomega.Equal([]interface{}{nil}), "Job run %s shouldn't have errors", jr.ID)
+		}
+	}, "3m", "3s").Should(gomega.Succeed())
+
+}

integration-tests/smoke/keeper_test.go

@@ -1126,3 +1126,69 @@ func setupKeeperTest(t *testing.T) (
 
 	return env.EVMClient, env.ClCluster.NodeAPIs(), env.ContractDeployer, linkTokenContract, env
 }
+
+func TestKeeperJobReplacement(t *testing.T) {
+	t.Parallel()
+	registryVersion := ethereum.RegistryVersion_1_3
+
+	l := logging.GetTestLogger(t)
+	chainClient, chainlinkNodes, contractDeployer, linkToken, _ := setupKeeperTest(t)
+	registry, _, consumers, upkeepIDs := actions.DeployKeeperContracts(
+		t,
+		registryVersion,
+		keeperDefaultRegistryConfig,
+		keeperDefaultUpkeepsToDeploy,
+		keeperDefaultUpkeepGasLimit,
+		linkToken,
+		contractDeployer,
+		chainClient,
+		big.NewInt(keeperDefaultLinkFunds),
+	)
+	gom := gomega.NewGomegaWithT(t)
+
+	_, err := actions.CreateKeeperJobsLocal(l, chainlinkNodes, registry, contracts.OCRv2Config{}, chainClient.GetChainID().String())
+	require.NoError(t, err, "Error creating keeper jobs")
+	err = chainClient.WaitForEvents()
+	require.NoError(t, err, "Error creating keeper jobs")
+
+	gom.Eventually(func(g gomega.Gomega) error {
+		// Check if the upkeeps are performing multiple times by analyzing their counters and checking they are greater than 10
+		for i := 0; i < len(upkeepIDs); i++ {
+			counter, err := consumers[i].Counter(utils.TestContext(t))
+			g.Expect(err).ShouldNot(gomega.HaveOccurred(), "Failed to retrieve consumer counter for upkeep at index %d", i)
+			g.Expect(counter.Int64()).Should(gomega.BeNumerically(">", int64(10)),
+				"Expected consumer counter to be greater than 10, but got %d", counter.Int64())
+			l.Info().Int64("Upkeep counter", counter.Int64()).Msg("Number of upkeeps performed")
+		}
+		return nil
+	}, "5m", "1s").Should(gomega.Succeed())
+
+	for _, n := range chainlinkNodes {
+		jobs, _, err := n.ReadJobs()
+		require.NoError(t, err)
+		for _, maps := range jobs.Data {
+			_, ok := maps["id"]
+			require.Equal(t, true, ok)
+			id := maps["id"].(string)
+			_, err := n.DeleteJob(id)
+			require.NoError(t, err)
+		}
+	}
+
+	_, err = actions.CreateKeeperJobsLocal(l, chainlinkNodes, registry, contracts.OCRv2Config{}, chainClient.GetChainID().String())
+	require.NoError(t, err, "Error creating keeper jobs")
+	err = chainClient.WaitForEvents()
+	require.NoError(t, err, "Error creating keeper jobs")
+
+	gom.Eventually(func(g gomega.Gomega) error {
+		// Check if the upkeeps are performing multiple times by analyzing their counters and checking they are greater than 10
+		for i := 0; i < len(upkeepIDs); i++ {
+			counter, err := consumers[i].Counter(utils.TestContext(t))
+			g.Expect(err).ShouldNot(gomega.HaveOccurred(), "Failed to retrieve consumer counter for upkeep at index %d", i)
+			g.Expect(counter.Int64()).Should(gomega.BeNumerically(">", int64(10)),
+				"Expected consumer counter to be greater than 10, but got %d", counter.Int64())
+			l.Info().Int64("Upkeep counter", counter.Int64()).Msg("Number of upkeeps performed")
+		}
+		return nil
+	}, "5m", "1s").Should(gomega.Succeed())
+}

integration-tests/smoke/keeper_test.go_test_list.json

@@ -53,6 +53,9 @@
     },
     {
       "name": "TestKeeperUpdateCheckData"
+    },
+    {
+      "name": "TestKeeperJobReplacement"
     }
   ]
 }

integration-tests/smoke/ocr2_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/smartcontractkit/chainlink-testing-framework/logging"
-
 	"github.com/smartcontractkit/chainlink/integration-tests/actions"
 	"github.com/smartcontractkit/chainlink/integration-tests/contracts"
 	"github.com/smartcontractkit/chainlink/integration-tests/docker/test_env"
@@ -92,3 +91,95 @@ func TestOCRv2Basic(t *testing.T) {
 		roundData.Answer.Int64(),
 	)
 }
+
+func TestOCRv2JobReplacement(t *testing.T) {
+	l := logging.GetTestLogger(t)
+
+	env, err := test_env.NewCLTestEnvBuilder().
+		WithTestLogger(t).
+		WithGeth().
+		WithMockAdapter().
+		WithCLNodeConfig(node.NewConfig(node.NewBaseConfig(),
+			node.WithOCR2(),
+			node.WithP2Pv2(),
+			node.WithTracing(),
+		)).
+		WithCLNodes(6).
+		WithFunding(big.NewFloat(.1)).
+		WithStandardCleanup().
+		Build()
+	require.NoError(t, err)
+
+	env.ParallelTransactions(true)
+
+	nodeClients := env.ClCluster.NodeAPIs()
+	bootstrapNode, workerNodes := nodeClients[0], nodeClients[1:]
+
+	linkToken, err := env.ContractDeployer.DeployLinkTokenContract()
+	require.NoError(t, err, "Deploying Link Token Contract shouldn't fail")
+
+	err = actions.FundChainlinkNodesLocal(workerNodes, env.EVMClient, big.NewFloat(.05))
+	require.NoError(t, err, "Error funding Chainlink nodes")
+
+	// Gather transmitters
+	var transmitters []string
+	for _, node := range workerNodes {
+		addr, err := node.PrimaryEthAddress()
+		if err != nil {
+			require.NoError(t, fmt.Errorf("error getting node's primary ETH address: %w", err))
+		}
+		transmitters = append(transmitters, addr)
+	}
+
+	ocrOffchainOptions := contracts.DefaultOffChainAggregatorOptions()
+	aggregatorContracts, err := actions.DeployOCRv2Contracts(1, linkToken, env.ContractDeployer, transmitters, env.EVMClient, ocrOffchainOptions)
+	require.NoError(t, err, "Error deploying OCRv2 aggregator contracts")
+
+	err = actions.CreateOCRv2JobsLocal(aggregatorContracts, bootstrapNode, workerNodes, env.MockAdapter, "ocr2", 5, env.EVMClient.GetChainID().Uint64(), false)
+	require.NoError(t, err, "Error creating OCRv2 jobs")
+
+	ocrv2Config, err := actions.BuildMedianOCR2ConfigLocal(workerNodes, ocrOffchainOptions)
+	require.NoError(t, err, "Error building OCRv2 config")
+
+	err = actions.ConfigureOCRv2AggregatorContracts(env.EVMClient, ocrv2Config, aggregatorContracts)
+	require.NoError(t, err, "Error configuring OCRv2 aggregator contracts")
+
+	err = actions.StartNewOCR2Round(1, aggregatorContracts, env.EVMClient, time.Minute*5, l)
+	require.NoError(t, err, "Error starting new OCR2 round")
+	roundData, err := aggregatorContracts[0].GetRound(utils.TestContext(t), big.NewInt(1))
+	require.NoError(t, err, "Getting latest answer from OCR contract shouldn't fail")
+	require.Equal(t, int64(5), roundData.Answer.Int64(),
+		"Expected latest answer from OCR contract to be 5 but got %d",
+		roundData.Answer.Int64(),
+	)
+
+	err = env.MockAdapter.SetAdapterBasedIntValuePath("ocr2", []string{http.MethodGet, http.MethodPost}, 10)
+	require.NoError(t, err)
+	err = actions.StartNewOCR2Round(2, aggregatorContracts, env.EVMClient, time.Minute*5, l)
+	require.NoError(t, err)
+
+	roundData, err = aggregatorContracts[0].GetRound(utils.TestContext(t), big.NewInt(2))
+	require.NoError(t, err, "Error getting latest OCR answer")
+	require.Equal(t, int64(10), roundData.Answer.Int64(),
+		"Expected latest answer from OCR contract to be 10 but got %d",
+		roundData.Answer.Int64(),
+	)
+
+	err = actions.DeleteJobs(nodeClients)
+	require.NoError(t, err)
+
+	err = actions.DeleteBridges(nodeClients)
+	require.NoError(t, err)
+
+	err = actions.CreateOCRv2JobsLocal(aggregatorContracts, bootstrapNode, workerNodes, env.MockAdapter, "ocr2", 15, env.EVMClient.GetChainID().Uint64(), false)
+	require.NoError(t, err, "Error creating OCRv2 jobs")
+
+	err = actions.StartNewOCR2Round(3, aggregatorContracts, env.EVMClient, time.Minute*3, l)
+	require.NoError(t, err, "Error starting new OCR2 round")
+	roundData, err = aggregatorContracts[0].GetRound(utils.TestContext(t), big.NewInt(3))
+	require.NoError(t, err, "Getting latest answer from OCR contract shouldn't fail")
+	require.Equal(t, int64(15), roundData.Answer.Int64(),
+		"Expected latest answer from OCR contract to be 15 but got %d",
+		roundData.Answer.Int64(),
+	)
+}