From 87e9607525300b121f747321990b3f94663eb7b4 Mon Sep 17 00:00:00 2001 From: Cedric Date: Fri, 18 Sep 2026 12:14:04 +0000 Subject: [PATCH] Restore framework library (#23776) --- .../framework/capabilities_registry.go | 125 +++++ .../integration_tests/framework/don.go | 459 ++++++++++++++++++ .../framework/don_configuration.go | 64 +++ .../integration_tests/framework/ethereum.go | 76 +++ .../framework/fake_billing_client.go | 33 ++ .../framework/fake_dispatcher.go | 275 +++++++++++ .../framework/fake_libocr.go | 335 +++++++++++++ .../framework/fake_target.go | 93 ++++ .../framework/fake_trigger.go | 154 ++++++ .../integration_tests/framework/peer.go | 140 ++++++ .../framework/workflow_registry.go | 146 ++++++ 11 files changed, 1900 insertions(+) create mode 100644 core/capabilities/integration_tests/framework/capabilities_registry.go create mode 100644 core/capabilities/integration_tests/framework/don.go create mode 100644 core/capabilities/integration_tests/framework/don_configuration.go create mode 100644 core/capabilities/integration_tests/framework/ethereum.go create mode 100644 core/capabilities/integration_tests/framework/fake_billing_client.go create mode 100644 core/capabilities/integration_tests/framework/fake_dispatcher.go create mode 100644 core/capabilities/integration_tests/framework/fake_libocr.go create mode 100644 core/capabilities/integration_tests/framework/fake_target.go create mode 100644 core/capabilities/integration_tests/framework/fake_trigger.go create mode 100644 core/capabilities/integration_tests/framework/peer.go create mode 100644 core/capabilities/integration_tests/framework/workflow_registry.go diff --git a/core/capabilities/integration_tests/framework/capabilities_registry.go b/core/capabilities/integration_tests/framework/capabilities_registry.go new file mode 100644 index 00000000000..c2200b1d985 --- /dev/null +++ b/core/capabilities/integration_tests/framework/capabilities_registry.go @@ -0,0 +1,125 @@ +package framework + +import ( + "context" + "testing" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/accounts/abi/bind" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" + kcr "github.com/smartcontractkit/chainlink-evm/gethwrappers/keystone/generated/capabilities_registry_1_1_0" + p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" +) + +type CapabilitiesRegistry struct { + t *testing.T + backend *EthBlockchain + contract *kcr.CapabilitiesRegistry + addr common.Address + nodeOperatorID uint32 + nextDonID int +} + +func NewCapabilitiesRegistry(ctx context.Context, t *testing.T, backend *EthBlockchain) *CapabilitiesRegistry { + addr, _, contract, err := kcr.DeployCapabilitiesRegistry(backend.transactionOpts, backend.Client()) + require.NoError(t, err) + backend.Commit() + + _, err = contract.AddNodeOperators(backend.transactionOpts, []kcr.CapabilitiesRegistryNodeOperator{ + { + Admin: backend.transactionOpts.From, + Name: "TEST_NODE_OPERATOR", + }, + }) + require.NoError(t, err) + blockHash := backend.Commit() + + logs, err := backend.Client().FilterLogs(ctx, ethereum.FilterQuery{ + BlockHash: &blockHash, + FromBlock: nil, + ToBlock: nil, + Addresses: nil, + Topics: nil, + }) + + require.NoError(t, err) + + recLog, err := contract.ParseNodeOperatorAdded(logs[0]) + require.NoError(t, err) + + nopID := recLog.NodeOperatorId + + return &CapabilitiesRegistry{t: t, addr: addr, contract: contract, backend: backend, nodeOperatorID: nopID} +} + +func (r *CapabilitiesRegistry) getAddress() common.Address { + return r.addr +} + +type capability struct { + donCapabilityConfig *pb.CapabilityConfig + registryConfig kcr.CapabilitiesRegistryCapability + // internalOnly is true if the capability is published in the registry but not made available outside the DON in which it runs + internalOnly bool +} + +// SetupDON sets up a new DON with the given capabilities and returns the DON ID +func (r *CapabilitiesRegistry) setupDON(donInfo DonConfiguration, capabilities []capability) int { + hashedCapabilityIDs := make([][32]byte, 0, len(capabilities)) + + for _, c := range capabilities { + id, err := r.contract.GetHashedCapabilityId(&bind.CallOpts{}, c.registryConfig.LabelledName, c.registryConfig.Version) + require.NoError(r.t, err) + hashedCapabilityIDs = append(hashedCapabilityIDs, id) + } + + registryCapabilities := make([]kcr.CapabilitiesRegistryCapability, 0, len(capabilities)) + for _, c := range capabilities { + registryCapabilities = append(registryCapabilities, c.registryConfig) + } + + _, err := r.contract.AddCapabilities(r.backend.transactionOpts, registryCapabilities) + require.NoError(r.t, err) + + r.backend.Commit() + + peerIDs := make([][32]byte, 0, len(donInfo.p2pKeys)) + nodes := make([]kcr.CapabilitiesRegistryNodeParams, 0, len(donInfo.p2pKeys)) + for i, p2pkey := range donInfo.p2pKeys { + signer, innerErr := getSignerStringFromOCRKeyBundle(donInfo.KeyBundles[i]) + require.NoError(r.t, innerErr) + peer := peerIDAndOCRSigner{PeerID: p2ptypes.PeerID(p2pkey.PeerID()), Signer: signer} + peerIDs = append(peerIDs, p2pkey.PeerID()) + n, innerErr := peerToNode(r.nodeOperatorID, peer) + require.NoError(r.t, innerErr) + + n.HashedCapabilityIds = hashedCapabilityIDs + nodes = append(nodes, n) + } + + _, err = r.contract.AddNodes(r.backend.transactionOpts, nodes) + require.NoError(r.t, err) + r.backend.Commit() + + capabilityConfigurations := make([]kcr.CapabilitiesRegistryCapabilityConfiguration, 0, len(capabilities)) + for i, c := range capabilities { + configBinary, err2 := proto.Marshal(c.donCapabilityConfig) + require.NoError(r.t, err2) + + capabilityConfigurations = append(capabilityConfigurations, kcr.CapabilitiesRegistryCapabilityConfiguration{ + CapabilityId: hashedCapabilityIDs[i], + Config: configBinary, + }) + } + + _, err = r.contract.AddDON(r.backend.transactionOpts, peerIDs, capabilityConfigurations, true, donInfo.AcceptsWorkflows, donInfo.F) + require.NoError(r.t, err) + r.backend.Commit() + + r.nextDonID++ + return r.nextDonID +} diff --git a/core/capabilities/integration_tests/framework/don.go b/core/capabilities/integration_tests/framework/don.go new file mode 100644 index 00000000000..45a949a7c2c --- /dev/null +++ b/core/capabilities/integration_tests/framework/don.go @@ -0,0 +1,459 @@ +package framework + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "maps" + "strconv" + "testing" + "time" + + "github.com/ethereum/go-ethereum/accounts/abi/bind" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/ethkey" + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/ocr2key" + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/workflowkey" + commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" + "github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry" + commonconfig "github.com/smartcontractkit/chainlink-common/pkg/config" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services" + "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" + kcr "github.com/smartcontractkit/chainlink-evm/gethwrappers/keystone/generated/capabilities_registry_1_1_0" + "github.com/smartcontractkit/chainlink-evm/pkg/assets" + evmtestutils "github.com/smartcontractkit/chainlink-evm/pkg/testutils" + remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" + "github.com/smartcontractkit/chainlink/v2/core/internal/cltest" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" + "github.com/smartcontractkit/chainlink/v2/core/services/chainlink" + "github.com/smartcontractkit/chainlink/v2/core/services/job" + p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" + "github.com/smartcontractkit/chainlink/v2/core/services/standardcapabilities" + syncer "github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncer/v2" + wftypes "github.com/smartcontractkit/chainlink/v2/core/services/workflows/types" + "github.com/smartcontractkit/chainlink/v2/core/utils/testutils/heavyweight" +) + +type DonContext struct { + EthBlockchain *EthBlockchain + p2pNetwork *FakeRageP2PNetwork + capabilityRegistry *CapabilitiesRegistry + workflowRegistry *WorkflowRegistry + syncerFetcherFunc wftypes.FetcherFunc +} + +func CreateDonContext(ctx context.Context, t *testing.T) DonContext { + ethBlockchain := NewEthBlockchain(t, 1000, 1*time.Second) + rageP2PNetwork := NewFakeRageP2PNetwork(ctx, t, 1000) + capabilitiesRegistry := NewCapabilitiesRegistry(ctx, t, ethBlockchain) + + servicetest.Run(t, rageP2PNetwork) + servicetest.Run(t, ethBlockchain) + return DonContext{EthBlockchain: ethBlockchain, p2pNetwork: rageP2PNetwork, capabilityRegistry: capabilitiesRegistry} +} + +func CreateDonContextWithWorkflowRegistry(ctx context.Context, t *testing.T, syncerFetcherFunc wftypes.FetcherFunc) DonContext { + donContext := CreateDonContext(ctx, t) + workflowRegistry := NewWorkflowRegistry(ctx, t, donContext.EthBlockchain) + donContext.workflowRegistry = workflowRegistry + donContext.syncerFetcherFunc = syncerFetcherFunc + return donContext +} + +func (c DonContext) WaitForCapabilitiesToBeExposed(t *testing.T, dons ...*DON) { + allExpectedCapabilities := make(map[CapabilityRegistration]bool) + for _, don := range dons { + caps, err := don.GetExternalCapabilities() + require.NoError(t, err) + maps.Copy(allExpectedCapabilities, caps) + } + + require.Eventually(t, func() bool { + registrations := c.p2pNetwork.GetCapabilityRegistrations() + + for k := range allExpectedCapabilities { + if _, ok := registrations[k]; !ok { + return false + } + } + + return true + }, 1*time.Minute, 1*time.Second, "timeout waiting for capabilities to be exposed") +} + +func (c DonContext) WaitForWorkflowRegistryMetadata(t *testing.T, workflowName string, owner string, workflowID [32]byte) { + require.Eventually(t, func() bool { + wf, err := c.workflowRegistry.contract.GetWorkflowMetadata(&bind.CallOpts{}, common.HexToAddress(owner), workflowName) + if err != nil { + return false + } + return wf.WorkflowID == workflowID + }, 1*time.Minute, 5*time.Second, "timeout waiting for workflow") +} + +type capabilityNode struct { + *cltest.TestApplication + registry *registry.Registry + key ethkey.KeyV2 + KeyBundle ocr2key.KeyBundle + peer peerIDAndOCRSigner + workflowKey *workflowkey.Key + start func() +} + +type DON struct { + services.StateMachine + t *testing.T + id *uint32 + config DonConfiguration + lggr logger.Logger + nodes []*capabilityNode + standardCapabilityJobs []*job.Job + publishedCapabilities []capability + + initialised bool + + capabilitiesRegistry *CapabilitiesRegistry + workflowRegistry *WorkflowRegistry + + nodeConfigModifiers []func(c *chainlink.Config, node *capabilityNode) + + fakeLibOcr *FakeLibOCR + + triggerFactories []TriggerFactory + targetFactories []TargetFactory +} + +func NewDON(ctx context.Context, t *testing.T, lggr logger.Logger, donConfig DonConfiguration, + dependentDONs []commoncap.DON, donContext DonContext, supportsOCR bool, protocolRoundInterval time.Duration) *DON { + don := &DON{t: t, lggr: logger.Named(lggr, donConfig.name), config: donConfig, capabilitiesRegistry: donContext.capabilityRegistry, + workflowRegistry: donContext.workflowRegistry} + + if supportsOCR { + // This is required to support the non standard OCR3 capability - will be removed when required OCR3 behaviour is implemented as standard capabilities + don.fakeLibOcr = NewFakeLibOCR(t, lggr, donConfig.F, protocolRoundInterval) + servicetest.Run(t, don.fakeLibOcr) + } + + for i, member := range donConfig.Members { + dispatcher := donContext.p2pNetwork.NewDispatcherForNode(member) + capabilityRegistry := registry.NewRegistry(lggr) + + nodeInfo := commoncap.Node{ + PeerID: &member, + WorkflowDON: donConfig.DON, + CapabilityDONs: dependentDONs, + } + + signer, err := getSignerStringFromOCRKeyBundle(donConfig.KeyBundles[i]) + require.NoError(t, err) + cn := &capabilityNode{ + registry: capabilityRegistry, + key: donConfig.keys[i], + KeyBundle: donConfig.KeyBundles[i], + peer: peerIDAndOCRSigner{PeerID: member, Signer: signer}, + } + don.nodes = append(don.nodes, cn) + + var newOracleFactoryFn standardcapabilities.NewOracleFactoryFn + if supportsOCR { + factory := newFakeOracleFactoryFactory(t, lggr, donConfig.KeyBundles[i], len(donConfig.Members), donConfig.F, + protocolRoundInterval) + newOracleFactoryFn = factory.NewOracleFactory + } + + cn.start = func() { + node := startNewNode(ctx, t, logger.Named(lggr, donConfig.name+"-"+strconv.Itoa(i)), nodeInfo, donContext.EthBlockchain, + donContext.capabilityRegistry.getAddress(), dispatcher, + peerWrapper{peer: p2pPeer{member}}, capabilityRegistry, newOracleFactoryFn, + donConfig.keys[i], func(c *chainlink.Config) { + for _, modifier := range don.nodeConfigModifiers { + modifier(c, cn) + } + }, donContext.syncerFetcherFunc) + require.NoError(t, node.KeyStore.P2P().Add(ctx, donConfig.p2pKeys[i])) + workflowKeys, err := node.KeyStore.Workflow().GetAll() + require.NoError(t, err) + + // Workflow nodes should only have at most 1 workflow key. + require.LessOrEqual(t, len(workflowKeys), 1) + if len(workflowKeys) == 1 { + cn.workflowKey = &workflowKeys[0] + } + require.NoError(t, node.Start(t.Context())) + cn.TestApplication = node + } + } + + return don +} + +// Initialise must be called after all capabilities have been added to the DONs and before Start is called +func (d *DON) Initialise() { + id := d.capabilitiesRegistry.setupDON(d.config, d.publishedCapabilities) + + //nolint:gosec // disable G115 + d.config.ID = uint32(id) + d.id = &d.config.ID + + if d.config.AcceptsWorkflows && d.workflowRegistry != nil { + d.workflowRegistry.UpdateAllowedDons([]uint32{d.config.ID}) + d.nodeConfigModifiers = append(d.nodeConfigModifiers, func(c *chainlink.Config, node *capabilityNode) { + workflowRegistryAddressStr := d.workflowRegistry.addr.String() + c.Capabilities.WorkflowRegistry.Address = &workflowRegistryAddressStr + c.Capabilities.WorkflowRegistry.ChainID = new(fmt.Sprintf("%d", testutils.SimulatedChainID)) + }) + } + d.initialised = true +} + +func (d *DON) GetID() uint32 { + if d.config.ID == 0 { + panic("DON ID not set, call Initialise() first") + } + + return d.config.ID +} + +func (d *DON) GetExternalCapabilities() (map[CapabilityRegistration]bool, error) { + result := map[CapabilityRegistration]bool{} + for _, publishedCapability := range d.publishedCapabilities { + if publishedCapability.internalOnly { + continue + } + + for _, node := range d.nodes { + result[CapabilityRegistration{ + nodePeerID: hex.EncodeToString(node.peer.PeerID[:]), + capabilityID: publishedCapability.registryConfig.LabelledName + "@" + publishedCapability.registryConfig.Version, + capabilityDonID: d.GetID(), + }] = true + } + } + + return result, nil +} + +func (d *DON) GetConfigVersion() uint32 { + return d.config.ConfigVersion +} + +func (d *DON) GetF() uint8 { + return d.config.F +} + +func (d *DON) GetPeerIDsAndOCRSigners() []peerIDAndOCRSigner { + peers := make([]peerIDAndOCRSigner, 0, len(d.nodes)) + for _, node := range d.nodes { + peers = append(peers, node.peer) + } + return peers +} + +func (d *DON) GetWorkflowPublicKeys() []*[32]byte { + keys := make([]*[32]byte, 0, len(d.nodes)) + for _, node := range d.nodes { + if node.workflowKey != nil { + pubKey := node.workflowKey.PublicKey() + keys = append(keys, &pubKey) + } + } + return keys +} + +func (d *DON) Start(ctx context.Context) error { + for _, triggerFactory := range d.triggerFactories { + for _, node := range d.nodes { + trigger := triggerFactory.CreateNewTrigger(d.t) + if err := node.registry.Add(ctx, trigger); err != nil { + return fmt.Errorf("failed to add trigger: %w", err) + } + } + } + + for _, targetFactory := range d.targetFactories { + for _, node := range d.nodes { + target := targetFactory.CreateNewTarget(d.t) + if err := node.registry.Add(ctx, target); err != nil { + return fmt.Errorf("failed to add target: %w", err) + } + } + } + + for _, node := range d.nodes { + node.start() + } + + for _, capabilityJob := range d.standardCapabilityJobs { + if err := d.AddJob(ctx, capabilityJob); err != nil { + return fmt.Errorf("failed to add standard capability job: %w", err) + } + } + + return nil +} + +func (d *DON) Close() error { + for _, node := range d.nodes { + if err := node.Stop(); err != nil { + return fmt.Errorf("failed to stop node: %w", err) + } + } + + return nil +} + +const StandardCapabilityTemplateJobSpec = ` +type = "standardcapabilities" +schemaVersion = 1 +name = "%s" +command="%s" +config=%s +` + +func (d *DON) AddStandardCapability(name string, command string, config string) { + spec := fmt.Sprintf(StandardCapabilityTemplateJobSpec, name, command, config) + capabilitiesSpecJob, err := standardcapabilities.ValidatedStandardCapabilitiesSpec(spec) + require.NoError(d.t, err) + + d.standardCapabilityJobs = append(d.standardCapabilityJobs, &capabilitiesSpecJob) +} + +func (d *DON) AddPublishedStandardCapability(name string, command string, config string, + defaultCapabilityRequestConfig *pb.CapabilityConfig, + registryConfig kcr.CapabilitiesRegistryCapability) { + spec := fmt.Sprintf(StandardCapabilityTemplateJobSpec, name, command, config) + capabilitiesSpecJob, err := standardcapabilities.ValidatedStandardCapabilitiesSpec(spec) + require.NoError(d.t, err) + + d.standardCapabilityJobs = append(d.standardCapabilityJobs, &capabilitiesSpecJob) + + d.publishedCapabilities = append(d.publishedCapabilities, capability{ + donCapabilityConfig: defaultCapabilityRequestConfig, + registryConfig: registryConfig, + }) +} + +// TODO - add configuration for remote support - do this for each capability as an option +func (d *DON) AddTargetCapability(targetFactory TargetFactory) { + d.targetFactories = append(d.targetFactories, targetFactory) +} + +func (d *DON) AddTriggerCapability(triggerFactory TriggerFactory) { + d.triggerFactories = append(d.triggerFactories, triggerFactory) +} + +func (d *DON) AddJob(ctx context.Context, j *job.Job) error { + for _, node := range d.nodes { + err := node.AddJobV2(ctx, j) + if err != nil { + return fmt.Errorf("failed to add job: %w", err) + } + } + + return nil +} + +func (d *DON) AddWorkflow(workflow Workflow) error { + if !d.config.AcceptsWorkflows { + return errors.New("cannot add workflow to non-workflow DON") + } + + if !d.initialised { + return errors.New("cannot add workflow to non-initialised DON") + } + + d.workflowRegistry.RegisterWorkflow(workflow, *d.id) + + return nil +} + +func (d *DON) UpdateWorkflow(workflow UpdatedWorkflow) error { + if !d.config.AcceptsWorkflows { + return errors.New("cannot add workflow to non-workflow DON") + } + + if !d.initialised { + return errors.New("cannot add workflow to non-initialised DON") + } + + d.workflowRegistry.UpdateWorkflow(workflow, *d.id) + + return nil +} + +func (d *DON) ComputeHashKey(owner string, field string) [32]byte { + return d.workflowRegistry.ComputeHashKey(owner, field) +} + +type TriggerFactory interface { + CreateNewTrigger(t *testing.T) commoncap.TriggerCapability + GetTriggerID() string + GetTriggerName() string + GetTriggerVersion() string +} + +type TargetFactory interface { + CreateNewTarget(t *testing.T) commoncap.ExecutableCapability + GetTargetID() string + GetTargetName() string + GetTargetVersion() string +} + +func startNewNode(ctx context.Context, + t *testing.T, lggr logger.Logger, nodeInfo commoncap.Node, + ethBlockchain *EthBlockchain, capRegistryAddr common.Address, + dispatcher remotetypes.Dispatcher, + peerWrapper p2ptypes.PeerWrapper, + localCapabilities *registry.Registry, + newOracleFactoryFn standardcapabilities.NewOracleFactoryFn, + keyV2 ethkey.KeyV2, + setupCfg func(c *chainlink.Config), + fetcherFunc wftypes.FetcherFunc, +) *cltest.TestApplication { + config, _ := heavyweight.FullTestDBV2(t, func(c *chainlink.Config, s *chainlink.Secrets) { + c.Capabilities.ExternalRegistry.ChainID = new(fmt.Sprintf("%d", testutils.SimulatedChainID)) + c.Capabilities.ExternalRegistry.Address = new(capRegistryAddr.String()) + c.Capabilities.Peering.V2.Enabled = new(true) + c.Capabilities.WorkflowRegistry.SyncStrategy = new(syncer.SyncStrategyReconciliation) + c.Feature.FeedsManager = new(false) + c.Feature.LogPoller = new(true) + c.CRE.UseLocalTimeProvider = new(true) + c.CRE.EnableDKGRecipient = new(true) + + c.P2P.V2.DeltaDial = commonconfig.MustNewDuration(100 * time.Millisecond) + c.P2P.V2.DeltaReconcile = commonconfig.MustNewDuration(100 * time.Millisecond) + c.EVM[0].LogPollInterval = commonconfig.MustNewDuration(500 * time.Millisecond) + + if setupCfg != nil { + setupCfg(c) + } + }) + + n, err := ethBlockchain.Client().NonceAt(ctx, ethBlockchain.transactionOpts.From, nil) + require.NoError(t, err) + + tx := evmtestutils.NewLegacyTransaction( + n, keyV2.Address, + assets.Ether(1).ToInt(), + 21000, + assets.GWei(1).ToInt(), + nil) + signedTx, err := ethBlockchain.transactionOpts.Signer(ethBlockchain.transactionOpts.From, tx) + require.NoError(t, err) + err = ethBlockchain.Client().SendTransaction(ctx, signedTx) + require.NoError(t, err) + ethBlockchain.Commit() + + return cltest.NewApplicationWithConfigV2AndKeyOnSimulatedBlockchain(t, config, ethBlockchain.Backend, + nodeInfo, dispatcher, peerWrapper, newOracleFactoryFn, localCapabilities, keyV2, lggr, fetcherFunc, + NewFakeBillingClient()) +} + +// Functions below this point are for adding non-standard capabilities to a DON, deliberately verbose. Eventually these +// should be replaced with standard capabilities. diff --git a/core/capabilities/integration_tests/framework/don_configuration.go b/core/capabilities/integration_tests/framework/don_configuration.go new file mode 100644 index 00000000000..7d1bfe15156 --- /dev/null +++ b/core/capabilities/integration_tests/framework/don_configuration.go @@ -0,0 +1,64 @@ +package framework + +import ( + "errors" + "fmt" + + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/ethkey" + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/ocr2key" + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/p2pkey" + commoncap "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" +) + +type DonConfiguration struct { + commoncap.DON + name string + keys []ethkey.KeyV2 + KeyBundles []ocr2key.KeyBundle + p2pKeys []p2pkey.KeyV2 +} + +// NewDonConfigurationParams exists purely to make it obvious in the test code what DON configuration is being used +type NewDonConfigurationParams struct { + Name string + NumNodes int + F uint8 + AcceptsWorkflows bool +} + +func NewDonConfiguration(don NewDonConfigurationParams) (DonConfiguration, error) { + if don.NumNodes < int(3*don.F+1) { + return DonConfiguration{}, errors.New("invalid configuration, number of nodes must be at least 3*F+1") + } + + keyBundles, p2pKeys, err := getKeyBundlesAndP2PKeys(don.Name, don.NumNodes) + if err != nil { + return DonConfiguration{}, fmt.Errorf("failed to get key bundles and p2p keys: %w", err) + } + + donPeers := make([]p2ptypes.PeerID, len(p2pKeys)) + var donKeys []ethkey.KeyV2 + for i := range p2pKeys { + donPeers[i] = p2ptypes.PeerID(p2pKeys[i].PeerID()) + newKey, err := ethkey.NewV2() + if err != nil { + return DonConfiguration{}, fmt.Errorf("failed to create key: %w", err) + } + donKeys = append(donKeys, newKey) + } + + donConfiguration := DonConfiguration{ + DON: commoncap.DON{ + Members: donPeers, + F: don.F, + ConfigVersion: 1, + AcceptsWorkflows: don.AcceptsWorkflows, + }, + name: don.Name, + p2pKeys: p2pKeys, + keys: donKeys, + KeyBundles: keyBundles, + } + return donConfiguration, nil +} diff --git a/core/capabilities/integration_tests/framework/ethereum.go b/core/capabilities/integration_tests/framework/ethereum.go new file mode 100644 index 00000000000..9bfde22b5a3 --- /dev/null +++ b/core/capabilities/integration_tests/framework/ethereum.go @@ -0,0 +1,76 @@ +package framework + +import ( + "context" + "os" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/accounts/abi/bind" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/eth/ethconfig" + gethlog "github.com/ethereum/go-ethereum/log" + + "github.com/smartcontractkit/chainlink-common/pkg/services" + "github.com/smartcontractkit/chainlink-evm/pkg/assets" + "github.com/smartcontractkit/chainlink-evm/pkg/testutils" + evmtypes "github.com/smartcontractkit/chainlink-evm/pkg/types" + "github.com/smartcontractkit/chainlink/v2/core/internal/cltest" +) + +type EthBlockchain struct { + services.StateMachine + evmtypes.Backend + transactionOpts *bind.TransactOpts + + blockTimeProcessingTime time.Duration + + stopCh services.StopChan + wg sync.WaitGroup +} + +func NewEthBlockchain(t *testing.T, initialEth int, blockTimeProcessingTime time.Duration) *EthBlockchain { + transactOpts := testutils.MustNewSimTransactor(t) // config contract deployer and owner + genesisData := types.GenesisAlloc{transactOpts.From: {Balance: assets.Ether(initialEth).ToInt()}} + backend := cltest.NewSimulatedBackend(t, genesisData, ethconfig.Defaults.Miner.GasCeil) + gethlog.SetDefault(gethlog.NewLogger(gethlog.NewTerminalHandlerWithLevel(os.Stderr, gethlog.LevelWarn, true))) + backend.Commit() + + return &EthBlockchain{Backend: backend, stopCh: make(services.StopChan), + blockTimeProcessingTime: blockTimeProcessingTime, transactionOpts: transactOpts} +} + +func (b *EthBlockchain) Start(ctx context.Context) error { + return b.StartOnce("EthBlockchain", func() error { + b.wg.Go(func() { + ticker := time.NewTicker(b.blockTimeProcessingTime) + defer ticker.Stop() + + for { + select { + case <-b.stopCh: + return + case <-ctx.Done(): + return + case <-ticker.C: + b.Commit() + } + } + }) + + return nil + }) +} + +func (b *EthBlockchain) Close() error { + return b.StopOnce("EthBlockchain", func() error { + close(b.stopCh) + b.wg.Wait() + return nil + }) +} + +func (b *EthBlockchain) TransactionOpts() *bind.TransactOpts { + return b.transactionOpts +} diff --git a/core/capabilities/integration_tests/framework/fake_billing_client.go b/core/capabilities/integration_tests/framework/fake_billing_client.go new file mode 100644 index 00000000000..b34f521f098 --- /dev/null +++ b/core/capabilities/integration_tests/framework/fake_billing_client.go @@ -0,0 +1,33 @@ +package framework + +import ( + "context" + + "google.golang.org/protobuf/types/known/emptypb" + + billing "github.com/smartcontractkit/chainlink-protos/billing/go" + "github.com/smartcontractkit/chainlink/v2/core/services/workflows/metering" +) + +type fakeBillingClient struct { +} + +func NewFakeBillingClient() metering.BillingClient { + return &fakeBillingClient{} +} + +func (f fakeBillingClient) GetOrganizationCreditsByWorkflow(ctx context.Context, req *billing.GetOrganizationCreditsByWorkflowRequest) (*billing.GetOrganizationCreditsByWorkflowResponse, error) { + return &billing.GetOrganizationCreditsByWorkflowResponse{OrganizationId: "", Credits: &billing.OrganizationCredits{CreditsReserved: "", Credits: ""}}, nil +} + +func (f fakeBillingClient) GetWorkflowExecutionRates(context.Context, *billing.GetWorkflowExecutionRatesRequest) (*billing.GetWorkflowExecutionRatesResponse, error) { + return &billing.GetWorkflowExecutionRatesResponse{RateCards: []*billing.RateCard{{ResourceType: billing.ResourceType_RESOURCE_TYPE_COMPUTE, MeasurementUnit: billing.MeasurementUnit_MEASUREMENT_UNIT_MILLISECONDS, UnitsPerCredit: "0.0001"}}}, nil +} + +func (f fakeBillingClient) SubmitWorkflowReceipt(ctx context.Context, request *billing.SubmitWorkflowReceiptRequest) (*emptypb.Empty, error) { + return &emptypb.Empty{}, nil +} + +func (f fakeBillingClient) ReserveCredits(ctx context.Context, request *billing.ReserveCreditsRequest) (*billing.ReserveCreditsResponse, error) { + return &billing.ReserveCreditsResponse{Success: true, RateCards: []*billing.RateCard{{ResourceType: billing.ResourceType_RESOURCE_TYPE_COMPUTE, MeasurementUnit: billing.MeasurementUnit_MEASUREMENT_UNIT_MILLISECONDS, UnitsPerCredit: "0.0001"}}, Credits: "10000"}, nil +} diff --git a/core/capabilities/integration_tests/framework/fake_dispatcher.go b/core/capabilities/integration_tests/framework/fake_dispatcher.go new file mode 100644 index 00000000000..5cb5779af7c --- /dev/null +++ b/core/capabilities/integration_tests/framework/fake_dispatcher.go @@ -0,0 +1,275 @@ +package framework + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "maps" + "sync" + "testing" + "time" + + "google.golang.org/protobuf/proto" + + "github.com/smartcontractkit/libocr/ragep2p/types" + + "github.com/smartcontractkit/chainlink-common/pkg/services" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote" + remotetypes "github.com/smartcontractkit/chainlink/v2/core/capabilities/remote/types" + p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" +) + +// FakeRageP2PNetwork backs the dispatchers created for each node in the test and effectively +// acts as the rageP2P network layer. +type FakeRageP2PNetwork struct { + services.StateMachine + t *testing.T + readyError error + + chanBufferSize int + stopCh services.StopChan + wg sync.WaitGroup + + peerIDToBrokerNode map[p2ptypes.PeerID]*brokerNode + + capabilityRegistrations map[CapabilityRegistration]bool + + mux sync.Mutex +} + +func NewFakeRageP2PNetwork(ctx context.Context, t *testing.T, chanBufferSize int) *FakeRageP2PNetwork { + network := &FakeRageP2PNetwork{ + t: t, + stopCh: make(services.StopChan), + chanBufferSize: chanBufferSize, + peerIDToBrokerNode: make(map[p2ptypes.PeerID]*brokerNode), + capabilityRegistrations: make(map[CapabilityRegistration]bool), + } + + go func() { + <-ctx.Done() + network.SetReadyError(errors.New("context done")) + }() + + return network +} + +func (a *FakeRageP2PNetwork) Start(ctx context.Context) error { + return a.StartOnce("FakeRageP2PNetwork", func() error { + return nil + }) +} + +func (a *FakeRageP2PNetwork) Close() error { + return a.StopOnce("FakeRageP2PNetwork", func() error { + close(a.stopCh) + a.wg.Wait() + return nil + }) +} + +func (a *FakeRageP2PNetwork) Ready() error { + a.mux.Lock() + defer a.mux.Unlock() + return a.readyError +} + +func (a *FakeRageP2PNetwork) SetReadyError(err error) { + a.mux.Lock() + defer a.mux.Unlock() + a.readyError = err +} + +// NewDispatcherForNode creates a new dispatcher for a node with the given peer ID. +func (a *FakeRageP2PNetwork) NewDispatcherForNode(nodePeerID p2ptypes.PeerID) remotetypes.Dispatcher { + return &brokerDispatcher{ + callerPeerID: nodePeerID, + broker: a, + receivers: map[key]remotetypes.Receiver{}, + } +} + +func (a *FakeRageP2PNetwork) HealthReport() map[string]error { + return nil +} + +func (a *FakeRageP2PNetwork) Name() string { + return "FakeRageP2PNetwork" +} + +type CapabilityRegistration struct { + nodePeerID string + capabilityID string + capabilityDonID uint32 +} + +func (a *FakeRageP2PNetwork) GetCapabilityRegistrations() map[CapabilityRegistration]bool { + a.mux.Lock() + defer a.mux.Unlock() + + copiedRegistrations := make(map[CapabilityRegistration]bool) + maps.Copy(copiedRegistrations, a.capabilityRegistrations) + return copiedRegistrations +} + +func (a *FakeRageP2PNetwork) registerReceiverNode(nodePeerID p2ptypes.PeerID, capabilityID string, capabilityDonID uint32, receiver remotetypes.Receiver) { + a.mux.Lock() + defer a.mux.Unlock() + + a.capabilityRegistrations[CapabilityRegistration{ + nodePeerID: hex.EncodeToString(nodePeerID[:]), + capabilityID: capabilityID, + capabilityDonID: capabilityDonID, + }] = true + + node, ok := a.peerIDToBrokerNode[nodePeerID] + if !ok { + node = a.newNode() + a.peerIDToBrokerNode[nodePeerID] = node + } + + node.registerReceiverCh <- ®isterReceiverRequest{ + receiverKey: receiverKey{ + capabilityID: capabilityID, + donID: capabilityDonID, + }, + receiver: receiver, + } +} + +func (a *FakeRageP2PNetwork) Send(msg *remotetypes.MessageBody) { + peerID := toPeerID(msg.Receiver) + + node, ok := a.getNodeForPeerID(peerID) + if !ok { + panic(fmt.Sprintf("node not found for peer ID %v", peerID)) + } + + node.receiveCh <- msg +} + +func (a *FakeRageP2PNetwork) getNodeForPeerID(peerID types.PeerID) (*brokerNode, bool) { + a.mux.Lock() + defer a.mux.Unlock() + node, ok := a.peerIDToBrokerNode[peerID] + return node, ok +} + +type brokerNode struct { + registerReceiverCh chan *registerReceiverRequest + receiveCh chan *remotetypes.MessageBody +} + +type receiverKey struct { + capabilityID string + donID uint32 +} + +type registerReceiverRequest struct { + receiverKey + receiver remotetypes.Receiver +} + +func (a *FakeRageP2PNetwork) newNode() *brokerNode { + n := &brokerNode{ + receiveCh: make(chan *remotetypes.MessageBody, a.chanBufferSize), + registerReceiverCh: make(chan *registerReceiverRequest, a.chanBufferSize), + } + + a.wg.Go(func() { + receivers := make(map[receiverKey]remotetypes.Receiver) + for { + select { + case <-a.stopCh: + return + case msg := <-n.receiveCh: + k := receiverKey{ + capabilityID: msg.CapabilityId, + donID: msg.CapabilityDonId, + } + + r, ok := receivers[k] + if !ok { + panic(fmt.Sprintf("receiver not found for key %+v", k)) + } + + r.Receive(a.t.Context(), msg) + case reg := <-n.registerReceiverCh: + receivers[reg.receiverKey] = reg.receiver + } + } + }) + return n +} + +func toPeerID(id []byte) p2ptypes.PeerID { + return [32]byte(id) +} + +type broker interface { + Send(msg *remotetypes.MessageBody) + Ready() error +} + +type brokerDispatcher struct { + callerPeerID p2ptypes.PeerID + broker broker + + receivers map[key]remotetypes.Receiver + mu sync.Mutex +} + +type key struct { + capID string + donID uint32 +} + +func (t *brokerDispatcher) Send(peerID p2ptypes.PeerID, msgBody *remotetypes.MessageBody) error { + clonedMsg := proto.Clone(msgBody).(*remotetypes.MessageBody) + clonedMsg.Version = 1 + clonedMsg.Sender = t.callerPeerID[:] + clonedMsg.Receiver = peerID[:] + clonedMsg.Timestamp = time.Now().UnixMilli() + t.broker.Send(clonedMsg) + return nil +} + +func (t *brokerDispatcher) SetReceiver(capabilityID string, donID uint32, receiver remotetypes.Receiver) error { + t.mu.Lock() + defer t.mu.Unlock() + k := key{capID: capabilityID, donID: donID} + _, ok := t.receivers[k] + if ok { + return fmt.Errorf("%w: receiver already exists for capability %s and don %d", remote.ErrReceiverExists, capabilityID, donID) + } + t.receivers[k] = receiver + + t.broker.(*FakeRageP2PNetwork).registerReceiverNode(t.callerPeerID, capabilityID, donID, receiver) + return nil +} +func (t *brokerDispatcher) RemoveReceiver(capabilityID string, donID uint32) {} + +func (t *brokerDispatcher) SetReceiverForMethod(capabilityID string, donID uint32, method string, receiver remotetypes.Receiver) error { + return errors.New("not implemented") +} +func (t *brokerDispatcher) RemoveReceiverForMethod(capabilityID string, donID uint32, method string) { +} + +func (t *brokerDispatcher) Start(context.Context) error { return nil } + +func (t *brokerDispatcher) Close() error { + return nil +} + +func (t *brokerDispatcher) Ready() error { + return t.broker.Ready() +} + +func (t *brokerDispatcher) HealthReport() map[string]error { + return nil +} + +func (t *brokerDispatcher) Name() string { + return "fakeDispatcher" +} diff --git a/core/capabilities/integration_tests/framework/fake_libocr.go b/core/capabilities/integration_tests/framework/fake_libocr.go new file mode 100644 index 00000000000..7731584bafe --- /dev/null +++ b/core/capabilities/integration_tests/framework/fake_libocr.go @@ -0,0 +1,335 @@ +package framework + +import ( + "bytes" + "context" + "fmt" + "math/rand" + "sync" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/smartcontractkit/libocr/commontypes" + "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/ocr2key" + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/services" + coretypes "github.com/smartcontractkit/chainlink-common/pkg/types/core" + "github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/generic" +) + +type oracleContext struct { + t *testing.T + lggr logger.Logger + key ocr2key.KeyBundle + N int + F uint8 + protocolRoundInterval time.Duration + mux sync.Mutex + pluginNameToFakeOcr map[string]*FakeLibOCR +} + +func (m *oracleContext) addPlugin(ctx context.Context, info ocr3types.ReportingPluginInfo, plugin ocr3types.ReportingPlugin[[]byte], + args coretypes.OracleArgs) error { + m.mux.Lock() + defer m.mux.Unlock() + + libOcr := m.pluginNameToFakeOcr[info.Name] + if libOcr == nil { + libOcr = NewFakeLibOCR(m.t, m.lggr, m.F, m.protocolRoundInterval) + m.pluginNameToFakeOcr[info.Name] = libOcr + } + + libOcr.AddNode(plugin, args.ContractTransmitter, m.key) + + if libOcr.GetNodeCount() == m.N { + err := libOcr.Start(ctx) + if err != nil { + return fmt.Errorf("failed to start fake lib ocr: %w", err) + } + } + return nil +} + +func (m *oracleContext) Close() error { + m.mux.Lock() + defer m.mux.Unlock() + + for _, libOcr := range m.pluginNameToFakeOcr { + if err := libOcr.Close(); err != nil { + return fmt.Errorf("failed to close fake lib ocr: %w", err) + } + } + return nil +} + +type oracleFactoryFactory struct { + oracleContext *oracleContext +} + +func newFakeOracleFactoryFactory(t *testing.T, lggr logger.Logger, key ocr2key.KeyBundle, n int, f uint8, protocolRoundInterval time.Duration) *oracleFactoryFactory { + return &oracleFactoryFactory{ + oracleContext: &oracleContext{ + t: t, + lggr: lggr, + key: key, + N: n, + F: f, + protocolRoundInterval: protocolRoundInterval, + pluginNameToFakeOcr: make(map[string]*FakeLibOCR), + }, + } +} + +func (o *oracleFactoryFactory) NewOracleFactory(params generic.OracleFactoryParams) (coretypes.OracleFactory, error) { + return &fakeOracleFactory{o.oracleContext}, nil +} + +type fakeOracleFactory struct { + oracleContext *oracleContext +} + +func (m *fakeOracleFactory) NewOracle(ctx context.Context, args coretypes.OracleArgs) (coretypes.Oracle, error) { + return &fakeOracle{oracleContext: m.oracleContext, args: args}, nil +} + +type fakeOracle struct { + oracleContext *oracleContext + args coretypes.OracleArgs +} + +func (m *fakeOracle) Start(ctx context.Context) error { + plugin, info, err := m.args.ReportingPluginFactoryService.NewReportingPlugin(ctx, ocr3types.ReportingPluginConfig{ + F: int(m.oracleContext.F), + N: m.oracleContext.N, + }) + + if err != nil { + return fmt.Errorf("failed to create reporting plugin: %w", err) + } + + if err = m.oracleContext.addPlugin(ctx, info, plugin, m.args); err != nil { + return fmt.Errorf("failed to add plugin: %w", err) + } + + return nil +} + +func (m *fakeOracle) Close(ctx context.Context) error { + return m.oracleContext.Close() +} + +type libocrNode struct { + id string + ocr3types.ReportingPlugin[[]byte] + ocr3types.ContractTransmitter[[]byte] + key ocr2key.KeyBundle +} + +// FakeLibOCR is a fake libocr implementation for testing purposes that simulates libocr protocol rounds without having +// to setup the libocr network +type FakeLibOCR struct { + services.StateMachine + t *testing.T + lggr logger.Logger + + nodes []*libocrNode + f uint8 + protocolRoundInterval time.Duration + + seqNr uint64 + outcomeCtx ocr3types.OutcomeContext + + mux sync.Mutex + stopCh services.StopChan + wg sync.WaitGroup +} + +func NewFakeLibOCR(t *testing.T, lggr logger.Logger, f uint8, protocolRoundInterval time.Duration) *FakeLibOCR { + return &FakeLibOCR{ + t: t, + lggr: lggr, + f: f, outcomeCtx: ocr3types.OutcomeContext{ + SeqNr: 1, + PreviousOutcome: nil, + }, + protocolRoundInterval: protocolRoundInterval, + stopCh: make(services.StopChan), + } +} + +func (m *FakeLibOCR) Start(ctx context.Context) error { + return m.StartOnce("FakeLibOCR", func() error { + m.wg.Go(func() { + ticker := time.NewTicker(m.protocolRoundInterval) + defer ticker.Stop() + + for { + select { + case <-m.stopCh: + return + case <-ticker.C: + serviceCtx, cancel := m.stopCh.NewCtx() + err := m.simulateProtocolRound(serviceCtx) + cancel() + if err != nil { + m.lggr.Errorf("simulating protocol round: %v", err) + } + } + } + }) + return nil + }) +} + +func (m *FakeLibOCR) Close() error { + return m.StopOnce("FakeLibOCR", func() error { + close(m.stopCh) + m.wg.Wait() + return nil + }) +} + +func (m *FakeLibOCR) AddNode(plugin ocr3types.ReportingPlugin[[]byte], transmitter ocr3types.ContractTransmitter[[]byte], key ocr2key.KeyBundle) string { + m.mux.Lock() + defer m.mux.Unlock() + node := &libocrNode{uuid.New().String(), plugin, transmitter, key} + m.nodes = append(m.nodes, node) + return node.id +} + +func (m *FakeLibOCR) GetNodeCount() int { + m.mux.Lock() + defer m.mux.Unlock() + return len(m.nodes) +} + +func (m *FakeLibOCR) RemoveNode(id string) { + m.mux.Lock() + defer m.mux.Unlock() + + var updatedNodes []*libocrNode + for _, node := range m.nodes { + if node.id != id { + updatedNodes = append(updatedNodes, node) + } + } + + m.nodes = updatedNodes +} + +func (m *FakeLibOCR) simulateProtocolRound(ctx context.Context) error { + m.mux.Lock() + defer m.mux.Unlock() + if len(m.nodes) == 0 { + return nil + } + + // randomly select a leader + leader := m.nodes[rand.Intn(len(m.nodes))] + + // get the query + query, err := leader.Query(ctx, m.outcomeCtx) + if err != nil { + return fmt.Errorf("failed to get query: %w", err) + } + + var observations []types.AttributedObservation + for oracleID, node := range m.nodes { + obs, err2 := node.Observation(ctx, m.outcomeCtx, query) + if err2 != nil { + return fmt.Errorf("failed to get observation: %w", err2) + } + + observations = append(observations, types.AttributedObservation{ + Observation: obs, + Observer: commontypes.OracleID(oracleID), + }) + } + + var outcomes []ocr3types.Outcome + for _, node := range m.nodes { + outcome, err2 := node.Outcome(ctx, m.outcomeCtx, query, observations) + if err2 != nil { + return fmt.Errorf("failed to get outcome: %w", err2) + } + + if len(outcome) == 0 { + return nil // wait until all nodes have an outcome for testing purposes + } + + outcomes = append(outcomes, outcome) + } + + // if all outcomes are equal proceed to reports + for _, outcome := range outcomes { + if !bytes.Equal(outcome, outcomes[0]) { + return nil + } + } + + reports, err := leader.Reports(ctx, 0, outcomes[0]) + if err != nil { + return fmt.Errorf("failed to get reports: %w", err) + } + for _, report := range reports { + // create signatures + var signatures []types.AttributedOnchainSignature + for i, node := range m.nodes { + sig, err := node.key.Sign(types.ReportContext{}, report.ReportWithInfo.Report) + if err != nil { + return fmt.Errorf("failed to sign report: %w", err) + } + + signatures = append(signatures, types.AttributedOnchainSignature{ + Signer: commontypes.OracleID(i), + Signature: sig, + }) + } + + for _, node := range m.nodes { + accept, err := node.ShouldAcceptAttestedReport(ctx, m.seqNr, report.ReportWithInfo) + if err != nil { + return fmt.Errorf("failed to check if report should be accepted: %w", err) + } + if !accept { + continue + } + + transmit, err := node.ShouldTransmitAcceptedReport(ctx, m.seqNr, report.ReportWithInfo) + if err != nil { + return fmt.Errorf("failed to check if report should be transmitted: %w", err) + } + + if !transmit { + continue + } + + // For each node select a random set of F+1 signatures to mimic libocr behaviour + s := rand.NewSource(time.Now().UnixNano()) + r := rand.New(s) + indices := r.Perm(len(signatures)) + selectedSignatures := make([]types.AttributedOnchainSignature, m.f+1) + for i := range int(m.f + 1) { + selectedSignatures[i] = signatures[indices[i]] + } + + err = node.Transmit(ctx, types.ConfigDigest{}, 0, report.ReportWithInfo, selectedSignatures) + if err != nil { + return fmt.Errorf("failed to transmit report: %w", err) + } + } + + m.seqNr++ + m.outcomeCtx = ocr3types.OutcomeContext{ + SeqNr: 0, + PreviousOutcome: outcomes[0], + } + } + + return nil +} diff --git a/core/capabilities/integration_tests/framework/fake_target.go b/core/capabilities/integration_tests/framework/fake_target.go new file mode 100644 index 00000000000..ed73a90f2df --- /dev/null +++ b/core/capabilities/integration_tests/framework/fake_target.go @@ -0,0 +1,93 @@ +package framework + +import ( + "context" + "testing" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/services" +) + +var ( + _ capabilities.ExecutableCapability = &fakeTarget{} +) + +type TargetSink struct { + services.StateMachine + targetID string + targetName string + version string + + targets []fakeTarget + Sink chan capabilities.CapabilityRequest +} + +func NewTargetSink(targetName string, version string) *TargetSink { + return &TargetSink{ + targetID: targetName + "@" + version, + targetName: targetName, + version: version, + Sink: make(chan capabilities.CapabilityRequest, 1000), + } +} + +func (ts *TargetSink) GetTargetVersion() string { + return ts.version +} + +func (ts *TargetSink) GetTargetName() string { + return ts.targetName +} + +func (ts *TargetSink) GetTargetID() string { + return ts.targetID +} + +func (ts *TargetSink) Start(ctx context.Context) error { + return ts.StartOnce("TargetSinkService", func() error { + return nil + }) +} + +func (ts *TargetSink) Close() error { + return ts.StopOnce("TargetSinkService", func() error { + return nil + }) +} + +func (ts *TargetSink) CreateNewTarget(t *testing.T) capabilities.ExecutableCapability { + target := fakeTarget{ + t: t, + targetID: ts.targetID, + ch: ts.Sink, + } + ts.targets = append(ts.targets, target) + return &target +} + +type fakeTarget struct { + t *testing.T + targetID string + ch chan capabilities.CapabilityRequest +} + +func (mt *fakeTarget) Execute(ctx context.Context, rawRequest capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error) { + mt.ch <- rawRequest + return capabilities.CapabilityResponse{}, nil +} + +func (mt *fakeTarget) Info(ctx context.Context) (capabilities.CapabilityInfo, error) { + return capabilities.MustNewCapabilityInfo( + mt.targetID, + capabilities.CapabilityTypeTarget, + "fake target for target ID "+mt.targetID, + ), nil +} + +func (mt *fakeTarget) RegisterToWorkflow(ctx context.Context, request capabilities.RegisterToWorkflowRequest) error { + return nil +} + +func (mt *fakeTarget) UnregisterFromWorkflow(ctx context.Context, request capabilities.UnregisterFromWorkflowRequest) error { + return nil +} diff --git a/core/capabilities/integration_tests/framework/fake_trigger.go b/core/capabilities/integration_tests/framework/fake_trigger.go new file mode 100644 index 00000000000..901856476db --- /dev/null +++ b/core/capabilities/integration_tests/framework/fake_trigger.go @@ -0,0 +1,154 @@ +package framework + +import ( + "context" + "sync" + "testing" + + "github.com/smartcontractkit/chainlink-common/pkg/capabilities" + "github.com/smartcontractkit/chainlink-common/pkg/services" + "github.com/smartcontractkit/chainlink-common/pkg/services/servicetest" + "github.com/smartcontractkit/chainlink-protos/cre/go/values" +) + +// TriggerSink is a TriggerFactory implementation that sends output to all triggers that are created by it. +type TriggerSink struct { + services.StateMachine + triggerID string + triggerName string + version string + + triggers []fakeTrigger + + stopCh services.StopChan + wg sync.WaitGroup +} + +func NewTriggerSink(t *testing.T, triggerName string, version string) *TriggerSink { + triggersFactory := &TriggerSink{ + triggerID: triggerName + "@" + version, + triggerName: triggerName, + version: version, + stopCh: make(services.StopChan), + } + servicetest.Run(t, triggersFactory) + return triggersFactory +} + +func (r *TriggerSink) GetTriggerVersion() string { + return r.version +} + +func (r *TriggerSink) GetTriggerName() string { + return r.triggerName +} + +func (r *TriggerSink) GetTriggerID() string { + return r.triggerID +} + +func (r *TriggerSink) Start(ctx context.Context) error { + return r.StartOnce("TriggerSink", func() error { + return nil + }) +} + +func (r *TriggerSink) Close() error { + return r.StopOnce("TriggerSink", func() error { + close(r.stopCh) + r.wg.Wait() + return nil + }) +} + +// SendOutput wraps the given output in a TriggerEvent and sends it to all triggers created by this factory +func (r *TriggerSink) SendOutput(outputs *values.Map, eventID string) { + triggerEvent := capabilities.TriggerEvent{ + TriggerType: r.triggerID, + ID: eventID, + Outputs: outputs, + } + + resp := capabilities.TriggerResponse{ + Event: triggerEvent, + } + + for _, trigger := range r.triggers { + trigger.sendResponse(resp) + } +} + +func (r *TriggerSink) CreateNewTrigger(t *testing.T) capabilities.TriggerCapability { + trigger := newFakeTrigger(t, r.triggerID, &r.wg, r.stopCh) + r.triggers = append(r.triggers, trigger) + return &trigger +} + +type fakeTrigger struct { + t *testing.T + triggerID string + cancel context.CancelFunc + toSend chan capabilities.TriggerResponse + + wg *sync.WaitGroup + stopCh services.StopChan +} + +func newFakeTrigger(t *testing.T, triggerID string, wg *sync.WaitGroup, stopCh services.StopChan) fakeTrigger { + return fakeTrigger{ + t: t, + triggerID: triggerID, + toSend: make(chan capabilities.TriggerResponse, 1000), + wg: wg, + stopCh: stopCh, + } +} + +func (s *fakeTrigger) sendResponse(resp capabilities.TriggerResponse) { + s.toSend <- resp +} + +func (s *fakeTrigger) Info(ctx context.Context) (capabilities.CapabilityInfo, error) { + return capabilities.MustNewCapabilityInfo( + s.triggerID, + capabilities.CapabilityTypeTrigger, + "fake trigger for trigger id "+s.triggerID, + ), nil +} + +func (s *fakeTrigger) AckEvent(ctx context.Context, triggerID string, eventID string, method string) error { + return nil +} + +func (s *fakeTrigger) RegisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) { + if s.cancel != nil { + s.t.Fatal("trigger already registered") + } + + responseCh := make(chan capabilities.TriggerResponse) + + ctxWithCancel, cancel := s.stopCh.NewCtx() + s.cancel = cancel + s.wg.Go(func() { + for { + select { + case <-ctxWithCancel.Done(): + return + case resp := <-s.toSend: + responseCh <- resp + } + } + }) + + return responseCh, nil +} + +func (s *fakeTrigger) UnregisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) error { + if s.cancel == nil { + s.t.Fatal("trigger not registered") + } + + s.cancel() + s.cancel = nil + return nil +} diff --git a/core/capabilities/integration_tests/framework/peer.go b/core/capabilities/integration_tests/framework/peer.go new file mode 100644 index 00000000000..84747e92889 --- /dev/null +++ b/core/capabilities/integration_tests/framework/peer.go @@ -0,0 +1,140 @@ +package framework + +import ( + "context" + "encoding/hex" + "errors" + "fmt" + "strings" + + "github.com/smartcontractkit/chainlink-common/keystore/corekeys" + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/ocr2key" + "github.com/smartcontractkit/chainlink-common/keystore/corekeys/p2pkey" + kcr "github.com/smartcontractkit/chainlink-evm/gethwrappers/keystone/generated/capabilities_registry_1_1_0" + "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" + p2ptypes "github.com/smartcontractkit/chainlink/v2/core/services/p2p/types" +) + +type peerIDAndOCRSigner struct { + PeerID p2ptypes.PeerID + Signer string +} + +func peerToNode(nopID uint32, p peerIDAndOCRSigner) (kcr.CapabilitiesRegistryNodeParams, error) { + sig := strings.TrimPrefix(p.Signer, "0x") + signerB, err := hex.DecodeString(sig) + if err != nil { + return kcr.CapabilitiesRegistryNodeParams{}, fmt.Errorf("failed to convert signer: %w", err) + } + + var sigb [32]byte + copy(sigb[:], signerB) + + return kcr.CapabilitiesRegistryNodeParams{ + NodeOperatorId: nopID, + P2pId: p.PeerID, + Signer: sigb, + EncryptionPublicKey: testutils.Random32Byte(), + }, nil +} + +func getKeyBundlesAndP2PKeys(donName string, numNodes int) ([]ocr2key.KeyBundle, []p2pkey.KeyV2, error) { + var keyBundles []ocr2key.KeyBundle + var donPeerKeys []p2pkey.KeyV2 + for range numNodes { + key, err := p2pkey.NewV2() + if err != nil { + return nil, nil, fmt.Errorf("failed to create p2p key: %w", err) + } + + keyBundle, err := ocr2key.New(corekeys.EVM) + if err != nil { + return nil, nil, fmt.Errorf("failed to create key bundle: %w", err) + } + + keyBundles = append(keyBundles, keyBundle) + + donPeerKeys = append(donPeerKeys, key) + } + return keyBundles, donPeerKeys, nil +} + +type peerWrapper struct { + peer p2pPeer +} + +func (t peerWrapper) Start(ctx context.Context) error { + return nil +} + +func (t peerWrapper) Close() error { + return nil +} + +func (t peerWrapper) Ready() error { + return nil +} + +func (t peerWrapper) HealthReport() map[string]error { + return nil +} + +func (t peerWrapper) Name() string { + return "peerWrapper" +} + +func (t peerWrapper) GetPeer() p2ptypes.Peer { + return t.peer +} + +type p2pPeer struct { + id p2ptypes.PeerID +} + +func (t p2pPeer) Start(ctx context.Context) error { + return nil +} + +func (t p2pPeer) Close() error { + return nil +} + +func (t p2pPeer) Ready() error { + return nil +} + +func (t p2pPeer) HealthReport() map[string]error { + return nil +} + +func (t p2pPeer) Name() string { + return "p2pPeer" +} + +func (t p2pPeer) ID() p2ptypes.PeerID { + return t.id +} + +func (t p2pPeer) UpdateConnections(peers map[p2ptypes.PeerID]p2ptypes.StreamConfig) error { + return nil +} + +func (t p2pPeer) Send(peerID p2ptypes.PeerID, msg []byte) error { + return nil +} + +func (t p2pPeer) Receive() <-chan p2ptypes.Message { + return nil +} + +func (t p2pPeer) IsBootstrap() bool { + return false +} + +func getSignerStringFromOCRKeyBundle(keyBundle ocr2key.KeyBundle) (string, error) { + if keyBundle == nil { + return "", errors.New("keyBundle is nil") + } + + return fmt.Sprintf("0x%x", keyBundle.PublicKey()), nil +} diff --git a/core/capabilities/integration_tests/framework/workflow_registry.go b/core/capabilities/integration_tests/framework/workflow_registry.go new file mode 100644 index 00000000000..abe7d44c821 --- /dev/null +++ b/core/capabilities/integration_tests/framework/workflow_registry.go @@ -0,0 +1,146 @@ +package framework + +import ( + "context" + "testing" + + "github.com/ethereum/go-ethereum/accounts/abi/bind" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/require" + + workflow_registry_wrapper "github.com/smartcontractkit/chainlink-evm/gethwrappers/workflow/generated/workflow_registry_wrapper_v1" +) + +type WorkflowRegistry struct { + t *testing.T + backend *EthBlockchain + contract *workflow_registry_wrapper.WorkflowRegistry + addr common.Address +} + +func NewWorkflowRegistry(ctx context.Context, t *testing.T, backend *EthBlockchain) *WorkflowRegistry { + // Deploy a test workflow_registry + wfRegistryAddr, _, wfRegistryC, err := workflow_registry_wrapper.DeployWorkflowRegistry(backend.transactionOpts, backend.Client()) + backend.Commit() + require.NoError(t, err) + + // setup contract state to allow the secrets to be updated + updateAuthorizedAddress(t, backend, wfRegistryC, []common.Address{backend.transactionOpts.From}, true) + + return &WorkflowRegistry{t: t, addr: wfRegistryAddr, contract: wfRegistryC, backend: backend} +} + +func (r *WorkflowRegistry) UpdateAllowedDons(donIDs []uint32) { + updateAllowedDONs(r.t, r.backend, r.contract, donIDs, true) +} + +func (r *WorkflowRegistry) RegisterWorkflow(input Workflow, donID uint32) { + registerWorkflow(r.t, r.backend, r.contract, input, donID) +} + +func (r *WorkflowRegistry) UpdateWorkflow(input UpdatedWorkflow, donID uint32) { + updateWorkflow(r.t, r.backend, r.contract, input) +} + +func (r *WorkflowRegistry) ComputeHashKey(owner string, field string) [32]byte { + return computeHashKey(r.t, r.contract, owner, field) +} + +func updateAuthorizedAddress( + t *testing.T, + th *EthBlockchain, + wfRegC *workflow_registry_wrapper.WorkflowRegistry, + addresses []common.Address, + _ bool, +) { + t.Helper() + _, err := wfRegC.UpdateAuthorizedAddresses(th.transactionOpts, addresses, true) + require.NoError(t, err, "failed to update authorised addresses") + th.Commit() + th.Commit() + th.Commit() + gotAddresses, err := wfRegC.GetAllAuthorizedAddresses(&bind.CallOpts{ + From: th.transactionOpts.From, + }) + require.NoError(t, err) + require.ElementsMatch(t, addresses, gotAddresses) +} + +func updateAllowedDONs( + t *testing.T, + th *EthBlockchain, + wfRegC *workflow_registry_wrapper.WorkflowRegistry, + donIDs []uint32, + allowed bool, +) { + t.Helper() + _, err := wfRegC.UpdateAllowedDONs(th.transactionOpts, donIDs, allowed) + require.NoError(t, err, "failed to update DONs") + th.Commit() + th.Commit() + th.Commit() + gotDons, err := wfRegC.GetAllAllowedDONs(&bind.CallOpts{ + From: th.transactionOpts.From, + }) + require.NoError(t, err) + require.ElementsMatch(t, donIDs, gotDons) +} + +type Workflow struct { + Name string + ID [32]byte + Status uint8 + BinaryURL string + ConfigURL string + SecretsURL string +} + +func registerWorkflow( + t *testing.T, + th *EthBlockchain, + wfRegC *workflow_registry_wrapper.WorkflowRegistry, + input Workflow, + donID uint32, +) { + t.Helper() + _, err := wfRegC.RegisterWorkflow(th.transactionOpts, input.Name, input.ID, donID, + input.Status, input.BinaryURL, input.ConfigURL, input.SecretsURL) + require.NoError(t, err, "failed to register workflow") + th.Commit() + th.Commit() + th.Commit() +} + +type UpdatedWorkflow struct { + WorkflowKey [32]byte + ID [32]byte + BinaryURL string + ConfigURL string + SecretsURL string +} + +func updateWorkflow( + t *testing.T, + th *EthBlockchain, + wfRegC *workflow_registry_wrapper.WorkflowRegistry, + input UpdatedWorkflow, +) { + t.Helper() + _, err := wfRegC.UpdateWorkflow(th.transactionOpts, input.WorkflowKey, input.ID, input.BinaryURL, input.ConfigURL, input.SecretsURL) + require.NoError(t, err, "failed to update workflow") + th.Commit() + th.Commit() + th.Commit() +} + +func computeHashKey( + t *testing.T, + wfRegC *workflow_registry_wrapper.WorkflowRegistry, + owner string, + field string, +) [32]byte { + t.Helper() + hashKey, err := wfRegC.ComputeHashKey(&bind.CallOpts{}, common.HexToAddress(owner), field) + require.NoError(t, err, "failed to compute hash key") + return hashKey +}