run_test.go

package capability

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"testing"
	"time"

	"github.com/jonboulle/clockwork"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/samber/do/v2"
	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/registry"
	"github.com/smartcontractkit/chainlink-common/pkg/loop"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
	"github.com/smartcontractkit/chainlink-common/pkg/logger"
	"github.com/smartcontractkit/chainlink-common/pkg/services"
	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
	"github.com/smartcontractkit/chainlink-common/x/config/markup/tomlmarkup"
)

// fake stands in for a generated capability server, which is what a real constructor returns. Only
// its type matters here: nothing calls it, because building it is as far as this step goes.
//
// The interfaces are embedded rather than implemented so that this says what a capability is - a
// service and a capability - without a page of methods saying it.
type fake struct {
	services.Service
	capabilities.ExecutableAndTriggerCapability

	// info is what the fake reports itself as, which is what the registrar serves and announces
	// it by.
	info capabilities.CapabilityInfo

	// started and closed record what the root service did with it. Written on the run's goroutine
	// and read on the test's, but only after the run has returned, so the channel that reports that
	// is what orders them.
	started bool
	closed  bool
}

func (f *fake) Info(context.Context) (capabilities.CapabilityInfo, error) { return f.info, nil }

func (f *fake) Methods() []Method { return nil }

// fakeID is what the fake registers and announces itself as.
const fakeID = "fake@1.0.0"

// newFake builds one with a real service behind it, since the health checker asks the capability
// its name and whether it is ready.
func newFake() *fake { return newFakeNamed(logger.Nop()) }

// newFakeNamed is a fake whose service is named from lggr, as a real capability's is from the logger
// it is given.
func newFakeNamed(lggr logger.Logger) *fake {
	f := &fake{}
	f.info, _ = capabilities.NewCapabilityInfo(fakeID, capabilities.CapabilityTypeCombined, "a fake")
	f.Service, _ = services.Config{
		Name:  "FakeCapability",
		Start: func(context.Context) error { f.started = true; return nil },
		Close: func() error { f.closed = true; return nil },
	}.NewServiceEngine(lggr)
	return f
}

// execute drives the real entry point with args of its own, and stops it once it is up.
//
// os.Args rather than a seam into the command tree, because that is what RunErr reads and what a
// binary is actually started with - including argv[0], which is what names the root command.
//
// A run blocks until its context is cancelled, so this waits until the binary is serving and then
// cancels - which is a test standing in for the signal an operator would send. A run that fails on
// the way up never gets there and reports that instead.
func execute(t *testing.T, lggr logger.Logger, ctor func(do.Injector) (Capability, error), args ...string) error {
	t.Helper()
	return executeWith(t, lggr, ctor, nil, args...)
}

// executeWith is execute, with opts given to RunErr.
func executeWith(t *testing.T, lggr logger.Logger, ctor func(do.Injector) (Capability, error), opts []Option, args ...string) error {
	t.Helper()

	// A free port rather than the default, since tests share a machine and run alongside each
	// other - two of them on 8080 would collide. It is also how this tells that the binary is up.
	port := 0
	if (slices.Contains(args, "run") || slices.Contains(args, "embed")) && !slices.Contains(args, "--http.port") {
		port = freePort(t)
		args = append(args, "--http.port", strconv.Itoa(port))

		// No --capabilities.proxy-url: it is optional, and leaving it out is the simpler run - the
		// registry then resolves only what this binary hosts, and dials nothing.
	}

	previous := os.Args
	t.Cleanup(func() { os.Args = previous })
	os.Args = append([]string{"cron"}, args...)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	done := make(chan error, 1)
	opts = append([]Option{withLogger(lggr), withRegisterer(prometheus.NewRegistry())}, opts...)
	go func() { done <- RunErr(ctx, "cron", ctor, opts...) }()

	if port == 0 {
		// Nothing that serves: a rejected constructor, or a bare invocation that prints help.
		return <-done
	}

	deadline := time.After(30 * time.Second)
	for {
		select {
		case err := <-done:
			return err
		case <-deadline:
			cancel()
			t.Fatal("the run never started serving")
			return nil
		case <-time.After(5 * time.Millisecond):
			if serving(port) {
				cancel()
				return <-done
			}
		}
	}
}

// serving checks /startupz because the web server answers before the run has finished starting.
//
// A timeout, because the port is not always the run's: a test that takes the port to watch the
// run fail leaves a listener that accepts and never answers, and a bare Get would hang on it
// rather than report not-serving.
func serving(port int) bool {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	res, err := httpGet(ctx, fmt.Sprintf("http://localhost:%d/startupz", port))
	if err != nil {
		return false
	}
	_ = res.Body.Close()
	return res.StatusCode == http.StatusOK
}

func httpGet(ctx context.Context, url string) (*http.Response, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}
	return http.DefaultClient.Do(req)
}

// freePort is a port nothing is listening on, found by listening on one and stopping again.
func freePort(t *testing.T) int {
	t.Helper()

	l, err := net.Listen("tcp", ":0")
	require.NoError(t, err)
	port := l.Addr().(*net.TCPAddr).Port
	require.NoError(t, l.Close())
	return port
}

// fakeConstructor builds a fake and asks the injector for nothing.
func fakeConstructor(do.Injector) (Capability, error) { return newFake(), nil }

func TestRunErrRejects(t *testing.T) {
	t.Run("no constructor", func(t *testing.T) {
		require.ErrorContains(t, RunErr[Capability](context.Background(), "cron", nil, withLogger(logger.Test(t))), "must provide a capability constructor")
	})

	t.Run("no app name", func(t *testing.T) {
		require.ErrorContains(t, RunErr(context.Background(), "", fakeConstructor, withLogger(logger.Test(t))), "must provide an app name")
	})
}

func TestRunBuildsTheCapability(t *testing.T) {
	built := 0

	require.NoError(t, execute(t, logger.Test(t), func(do.Injector) (Capability, error) {
		built++
		return newFake(), nil
	}, "run"))
	assert.Equal(t, 1, built, "the constructor should have been called exactly once")
}

func TestRunReportsAFailedConstructor(t *testing.T) {
	err := execute(t, logger.Test(t), func(do.Injector) (Capability, error) {
		return nil, errors.New("no schedule")
	}, "run")
	require.ErrorContains(t, err, "failed to instantiate capability: no schedule")
}

// TestRunRejectsANilCapability covers a constructor that reports success without building
// anything: the run says so rather than starting a nil capability.
func TestRunRejectsANilCapability(t *testing.T) {
	err := execute(t, logger.Test(t), func(do.Injector) (Capability, error) { return nil, nil }, "run")
	require.ErrorContains(t, err, "returned neither a capability nor an error")
}

// TestRunProvidesItsDependencies covers what the injector holds: this process's logger, the
// registry capabilities are resolved through, the factory a workflow's requests are bounded by, and
// the clock.
func TestRunProvidesItsDependencies(t *testing.T) {
	var (
		gotLogger   logger.Logger
		gotRegistry registry.CapabilitiesRegistry
		gotLimits   limits.Factory
		gotClock    clockwork.Clock
	)

	require.NoError(t, execute(t, logger.Test(t), func(i do.Injector) (Capability, error) {
		gotLogger = do.MustInvoke[logger.Logger](i)
		gotRegistry = do.MustInvoke[registry.CapabilitiesRegistry](i)
		gotLimits = do.MustInvoke[limits.Factory](i)
		gotClock = do.MustInvoke[clockwork.Clock](i)
		return newFake(), nil
	}, "run"))

	assert.NotNil(t, gotLogger)
	assert.NotNil(t, gotRegistry)
	assert.NotNil(t, gotLimits.Settings)
	assert.NotNil(t, gotLimits.Meter)
	assert.NotNil(t, gotClock)
}

// TestBareCommandRunsInCompatibilityMode covers a bare <command>, which is how the core node starts
// a LOOP. The test can't reach plugin.Serve, because it takes over the stdio of the process. So
// each case must fail before the capability is built.
func TestBareCommandRunsInCompatibilityMode(t *testing.T) {
	h := loop.EmptyHandshakeConfig()
	notBuilt := func(do.Injector) (Capability, error) {
		t.Error("the capability must not be built")
		return newFake(), nil
	}

	t.Run("without a plugin host", func(t *testing.T) {
		t.Setenv(h.MagicCookieKey, "")
		require.ErrorContains(t, execute(t, logger.Test(t), notBuilt), "must be started by a go-plugin host")
	})

	t.Run("without CL_PROMETHEUS_PORT", func(t *testing.T) {
		t.Setenv(h.MagicCookieKey, h.MagicCookieValue)
		require.ErrorContains(t, execute(t, logger.Test(t), notBuilt), "needs CL_PROMETHEUS_PORT")
	})
}

// TestBareCommandServesOnThePrometheusPort makes sure that compatibility mode serves on the port
// that the core node gives in CL_PROMETHEUS_PORT, and ignores CL_HTTP_PORT.
func TestBareCommandServesOnThePrometheusPort(t *testing.T) {
	h := loop.EmptyHandshakeConfig()
	promPort, httpPort := freePort(t), freePort(t)
	t.Setenv(h.MagicCookieKey, h.MagicCookieValue)
	t.Setenv("CL_PROMETHEUS_PORT", strconv.Itoa(promPort))
	t.Setenv("CL_HTTP_PORT", strconv.Itoa(httpPort))

	previous := os.Args
	t.Cleanup(func() { os.Args = previous })
	os.Args = []string{"cron"}

	served := false
	serve := func(context.Context, services.Service) {
		served = true
		assert.True(t, serving(promPort), "the web server must serve on CL_PROMETHEUS_PORT")
		assert.False(t, serving(httpPort), "the web server must not serve on CL_HTTP_PORT")
	}

	require.NoError(t, RunErr(t.Context(), "cron", fakeConstructor,
		withLogger(logger.Test(t)), withRegisterer(prometheus.NewRegistry()), withServe(serve)))
	assert.True(t, served)
	assertPortFree(t, promPort)
}

// TestRunServesAndAnnouncesTheCapability is the whole path through the real entry point: a run
// serves its capability, tells the node's registry where, and takes it back on shutdown.
//
// The removal half is reliable to assert here because the registry's close is one ordered
// function: the Remove RPC is sent before the connection it travels on is closed.
func TestRunServesAndAnnouncesTheCapability(t *testing.T) {
	node := newSpyRegistry(t)
	proxyURL := serveNodeRegistry(t, node)

	require.NoError(t, execute(t, logger.Test(t), fakeConstructor, "run", "--capabilities.proxy-url", proxyURL))

	assert.Equal(t, []string{fakeID}, node.addedIDs(), "the run should have announced the capability to the node's registry")
	assert.Equal(t, []string{fakeID}, node.removedIDs(), "and deregistered it on the way out")
}

// TestRunInjectorConfigIsOnlyOnTheRunCommand covers which command each injector's configs are registered on.
func TestRunInjectorConfigIsOnlyOnTheRunCommand(t *testing.T) {
	type runOnly struct{ Name string }
	conf := &runOnly{}

	runInjector := RunInjector
	t.Cleanup(func() { RunInjector = runInjector })
	RunInjector = runInjector.Clone()
	require.NoError(t, RunInjector.ProvideConfig("runonly", conf))

	require.NoError(t, execute(t, logger.Test(t), fakeConstructor, "run", "--runonly.name", "set"))
	assert.Equal(t, "set", conf.Name)

	require.ErrorContains(t, execute(t, logger.Test(t), fakeConstructor, "embed", "1", "--runonly.name", "set"),
		"unknown flag: --runonly.name")
}

// TestWithConfig covers the config that a capability asks for with WithConfig. The node passes each
// key as its own CL_<APP>_<KEY> env var. Each case runs compat without a plugin host, which fails
// after the binder decodes and before anything is served.
func TestWithConfig(t *testing.T) {
	type appConf struct {
		Name  string
		Count int
	}

	compatWith := func(t *testing.T, opts []Option, args ...string) (*appConf, error) {
		var runInjector *Injector
		capture := func(c *config) error {
			runInjector = c.runInjector
			return nil
		}
		opts = append([]Option{WithConfig[appConf](), capture}, opts...)
		err := executeWith(t, logger.Test(t), fakeConstructor, opts, append([]string{"compat"}, args...)...)
		require.NotNil(t, runInjector, "RunErr returned before it applied every option: %v", err)
		conf, invokeErr := do.Invoke[*appConf](runInjector.Do())
		require.NoError(t, invokeErr)
		return conf, err
	}

	compat := func(t *testing.T, args ...string) (*appConf, error) { return compatWith(t, nil, args...) }

	writeConfig := func(t *testing.T, name, contents string) string {
		path := filepath.Join(t.TempDir(), name)
		require.NoError(t, os.WriteFile(path, []byte(contents), 0o600))
		return path
	}

	t.Run("reads a YAML config file by default", func(t *testing.T) {
		path := writeConfig(t, "config.yaml", "cron:\n  Name: node\n  Count: 3\n")
		conf, err := compat(t, "--config", path)
		require.ErrorContains(t, err, "must be started by a go-plugin host")
		assert.Equal(t, appConf{Name: "node", Count: 3}, *conf)
	})

	t.Run("reads a config file in the markup given by WithMarkup", func(t *testing.T) {
		path := writeConfig(t, "config.toml", "[cron]\nName = 'node'\nCount = 3\n")
		conf, err := compatWith(t, []Option{WithMarkup(tomlmarkup.New())}, "--config", path)
		require.ErrorContains(t, err, "must be started by a go-plugin host")
		assert.Equal(t, appConf{Name: "node", Count: 3}, *conf)
	})

	t.Run("fills the config from env vars under the app name", func(t *testing.T) {
		t.Setenv("CL_CRON_NAME", "node")
		t.Setenv("CL_CRON_COUNT", "3")
		conf, err := compat(t)
		require.ErrorContains(t, err, "must be started by a go-plugin host")
		assert.Equal(t, appConf{Name: "node", Count: 3}, *conf)
	})

	t.Run("leaves the keys it is not given", func(t *testing.T) {
		t.Setenv("CL_CRON_COUNT", "3")
		conf, err := compat(t)
		require.ErrorContains(t, err, "must be started by a go-plugin host")
		assert.Equal(t, appConf{Count: 3}, *conf)
	})

	t.Run("is overridden by flags", func(t *testing.T) {
		t.Setenv("CL_CRON_NAME", "node")
		conf, err := compat(t, "--cron.name", "flag")
		require.ErrorContains(t, err, "must be started by a go-plugin host")
		assert.Equal(t, appConf{Name: "flag"}, *conf)
	})

	t.Run("ignores a TOML config in CL_CAPABILITIES_CONFIG and CRE_ env vars", func(t *testing.T) {
		t.Setenv("CL_CAPABILITIES_CONFIG", "[cron]\nName = 'node'\nCount = 3\n")
		t.Setenv("CRE_CRON_COUNT", "3")
		conf, err := compat(t)
		require.ErrorContains(t, err, "must be started by a go-plugin host")
		assert.Equal(t, appConf{}, *conf)
	})

	t.Run("rejects a value of the wrong type", func(t *testing.T) {
		t.Setenv("CL_CRON_COUNT", "three")
		_, err := compat(t)
		require.ErrorContains(t, err, "CL_CRON_COUNT")
	})

	t.Run("returns an error when given the same config twice", func(t *testing.T) {
		err := executeWith(t, logger.Test(t), fakeConstructor, []Option{WithConfig[appConf](), WithConfig[appConf]()}, "compat")
		require.ErrorContains(t, err, "is already provided")
	})
}