base 2 commit

.gitignore

@@ -73,3 +73,6 @@ __debug*
 .private_chains/*
 
 .direnv
+
+# asdf
+.tool-versions

lib/docker/test_env/job_distributor/jd.go

@@ -0,0 +1,193 @@
+package job_distributor
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+	tc "github.com/testcontainers/testcontainers-go"
+	tcwait "github.com/testcontainers/testcontainers-go/wait"
+
+	"github.com/goplugin/plugin-testing-framework/lib/docker"
+	"github.com/goplugin/plugin-testing-framework/lib/docker/test_env"
+	"github.com/goplugin/plugin-testing-framework/lib/logging"
+	"github.com/goplugin/plugin-testing-framework/lib/utils/testcontext"
+)
+
+const (
+	JDContainerName            string = "job-distributor"
+	DEAFULTJDContainerPort     string = "42242"
+	DEFAULTCSAKeyEncryptionKey string = "!PASsword000!"
+	DEAFULTWSRPCContainerPort  string = "8080"
+)
+
+type Option = func(j *Component)
+
+type Component struct {
+	test_env.EnvComponent
+	Grpc                string
+	Wsrpc               string
+	InternalGRPC        string
+	InternalWSRPC       string
+	l                   zerolog.Logger
+	t                   *testing.T
+	dbConnection        string
+	containerPort       string
+	wsrpcPort           string
+	csaKeyEncryptionKey string
+}
+
+func (j *Component) startOrRestartContainer(withReuse bool) error {
+	req := j.getContainerRequest()
+	l := logging.GetTestContainersGoTestLogger(j.t)
+	c, err := docker.StartContainerWithRetry(j.l, tc.GenericContainerRequest{
+		ContainerRequest: *req,
+		Started:          true,
+		Reuse:            withReuse,
+		Logger:           l,
+	})
+	if err != nil {
+		return err
+	}
+	j.Container = c
+	ctx := testcontext.Get(j.t)
+	host, err := test_env.GetHost(ctx, c)
+	if err != nil {
+		return errors.Wrapf(err, "cannot get host for container %s", j.ContainerName)
+	}
+
+	p, err := c.MappedPort(ctx, test_env.NatPort(j.containerPort))
+	if err != nil {
+		return errors.Wrapf(err, "cannot get container mapped port for container %s", j.ContainerName)
+	}
+	j.Grpc = fmt.Sprintf("%s:%s", host, p.Port())
+
+	p, err = c.MappedPort(ctx, test_env.NatPort(j.wsrpcPort))
+	if err != nil {
+		return errors.Wrapf(err, "cannot get wsrpc mapped port for container %s", j.ContainerName)
+	}
+	j.Wsrpc = fmt.Sprintf("%s:%s", host, p.Port())
+	j.InternalGRPC = fmt.Sprintf("%s:%s", j.ContainerName, j.containerPort)
+
+	j.InternalWSRPC = fmt.Sprintf("%s:%s", j.ContainerName, j.wsrpcPort)
+	j.l.Info().
+		Str("containerName", j.ContainerName).
+		Str("grpcURI", j.Grpc).
+		Str("wsrpcURI", j.Wsrpc).
+		Str("internalGRPC", j.InternalGRPC).
+		Str("internalWSRPC", j.InternalWSRPC).
+		Msg("Started Job Distributor container")
+
+	return nil
+}
+
+func (j *Component) getContainerRequest() *tc.ContainerRequest {
+	return &tc.ContainerRequest{
+		Name:  j.ContainerName,
+		Image: fmt.Sprintf("%s:%s", j.ContainerImage, j.ContainerVersion),
+		ExposedPorts: []string{
+			test_env.NatPortFormat(j.containerPort),
+			test_env.NatPortFormat(j.wsrpcPort),
+		},
+		Env: map[string]string{
+			"DATABASE_URL":              j.dbConnection,
+			"PORT":                      j.containerPort,
+			"NODE_RPC_PORT":             j.wsrpcPort,
+			"CSA_KEY_ENCRYPTION_SECRET": j.csaKeyEncryptionKey,
+		},
+		Networks: j.Networks,
+		WaitingFor: tcwait.ForAll(
+			tcwait.ForListeningPort(test_env.NatPort(j.containerPort)),
+			tcwait.ForListeningPort(test_env.NatPort(j.wsrpcPort)),
+		),
+		LifecycleHooks: []tc.ContainerLifecycleHooks{
+			{
+				PostStarts: j.PostStartsHooks,
+				PostStops:  j.PostStopsHooks,
+			},
+		},
+	}
+}
+
+func (j *Component) StartContainer() error {
+	return j.startOrRestartContainer(false)
+}
+
+func (j *Component) RestartContainer() error {
+	return j.startOrRestartContainer(true)
+}
+
+func New(networks []string, opts ...Option) *Component {
+	id, _ := uuid.NewRandom()
+	j := &Component{
+		EnvComponent: test_env.EnvComponent{
+			ContainerName:  fmt.Sprintf("%s-%s", JDContainerName, id.String()[0:8]),
+			Networks:       networks,
+			StartupTimeout: 2 * time.Minute,
+		},
+		containerPort:       DEAFULTJDContainerPort,
+		wsrpcPort:           DEAFULTWSRPCContainerPort,
+		csaKeyEncryptionKey: DEFAULTCSAKeyEncryptionKey,
+		l:                   log.Logger,
+	}
+	j.SetDefaultHooks()
+	for _, opt := range opts {
+		opt(j)
+	}
+	return j
+}
+
+func WithTestInstance(t *testing.T) Option {
+	return func(j *Component) {
+		j.l = logging.GetTestLogger(t)
+		j.t = t
+	}
+}
+
+func WithContainerPort(port string) Option {
+	return func(j *Component) {
+		j.containerPort = port
+	}
+}
+
+func WithWSRPCContainerPort(port string) Option {
+	return func(j *Component) {
+		j.wsrpcPort = port
+	}
+}
+
+func WithDBURL(db string) Option {
+	return func(j *Component) {
+		if db != "" {
+			j.dbConnection = db
+		}
+	}
+}
+
+func WithContainerName(name string) Option {
+	return func(j *Component) {
+		j.ContainerName = name
+	}
+}
+
+func WithImage(image string) Option {
+	return func(j *Component) {
+		j.ContainerImage = image
+	}
+}
+
+func WithVersion(version string) Option {
+	return func(j *Component) {
+		j.ContainerVersion = version
+	}
+}
+
+func WithCSAKeyEncryptionKey(key string) Option {
+	return func(j *Component) {
+		j.csaKeyEncryptionKey = key
+	}
+}

lib/docker/test_env/job_distributor/jd_test.go

@@ -0,0 +1,54 @@
+package job_distributor
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/goplugin/plugin-testing-framework/lib/docker"
+	"github.com/goplugin/plugin-testing-framework/lib/docker/test_env"
+	"github.com/goplugin/plugin-testing-framework/lib/logging"
+)
+
+func TestJDSpinUp(t *testing.T) {
+	t.Skipf("TODO enable this when jd image is available in ci")
+	l := logging.GetTestLogger(t)
+	network, err := docker.CreateNetwork(l)
+	require.NoError(t, err)
+
+	// create a postgres first
+	pg, err := test_env.NewPostgresDb(
+		[]string{network.Name},
+		test_env.WithPostgresDbName("jd-db"),
+		test_env.WithPostgresImageVersion("14.1"))
+	require.NoError(t, err)
+	err = pg.StartContainer()
+	require.NoError(t, err)
+
+	jd := New([]string{network.Name},
+		//TODO: replace with actual image
+		WithImage("localhost:5001/jd"),
+		WithVersion("latest"),
+		WithDBURL(pg.InternalURL.String()),
+	)
+
+	err = jd.StartContainer()
+	require.NoError(t, err)
+	// create a jd connection
+	_, err = newConnection(jd.Grpc)
+	require.NoError(t, err)
+}
+
+func newConnection(target string) (*grpc.ClientConn, error) {
+	var opts []grpc.DialOption
+	opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	conn, err := grpc.NewClient(target, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to connect to service at %s. Err: %w", target, err)
+	}
+
+	return conn, nil
+}

lib/docker/test_env/postgres.go

@@ -158,13 +158,13 @@ func (pg *PostgresDb) startOrRestartContainer(withReuse bool) error {
 	internalUrl, err := url.Parse(fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=disable",
 		pg.User, pg.Password, pg.ContainerName, "5432", pg.DbName))
 	if err != nil {
-		return fmt.Errorf("error parsing mercury db internal url: %w", err)
+		return errors.Wrap(err, "error parsing db internal url")
 	}
 	pg.InternalURL = internalUrl
 	externalUrl, err := url.Parse(fmt.Sprintf("postgres://%s:%[email protected]:%s/%s?sslmode=disable",
 		pg.User, pg.Password, externalPort.Port(), pg.DbName))
 	if err != nil {
-		return fmt.Errorf("error parsing mercury db external url: %w", err)
+		return errors.Wrap(err, "error parsing db external url")
 	}
 	pg.ExternalURL = externalUrl
 

lib/docker/test_env/utils.go

@@ -31,6 +31,21 @@ func GetHost(ctx context.Context, container tc.Container) (string, error) {
 	return host, nil
 }
 
+// GetEndpointFromPort returns the endpoint of a container associated with a port,
+// if localhost then force ipv4 localhost
+// to avoid ipv6 docker bugs https://github.com/moby/moby/issues/42442 https://github.com/moby/moby/issues/42375
+func GetEndpointFromPort(ctx context.Context, container tc.Container, endpointType string, portStr string) (string, error) {
+	port, err := nat.NewPort("tcp", portStr)
+	if err != nil {
+		return "", err
+	}
+	endpoint, err := container.PortEndpoint(ctx, port, endpointType)
+	if err != nil {
+		return "", err
+	}
+	return strings.Replace(endpoint, "localhost", "127.0.0.1", 1), nil
+}
+
 // GetEndpoint returns the endpoint of a container, if localhost then force ipv4 localhost
 // to avoid ipv6 docker bugs https://github.com/moby/moby/issues/42442 https://github.com/moby/moby/issues/42375
 func GetEndpoint(ctx context.Context, container tc.Container, endpointType string) (string, error) {

lib/go.mod

@@ -42,10 +42,10 @@ require (
 	github.com/testcontainers/testcontainers-go v0.28.0
 	go.uber.org/atomic v1.11.0
 	go.uber.org/zap v1.26.0
-	golang.org/x/net v0.26.0
+	golang.org/x/net v0.29.0
 	golang.org/x/oauth2 v0.21.0
-	golang.org/x/sync v0.7.0
-	golang.org/x/text v0.16.0
+	golang.org/x/sync v0.8.0
+	golang.org/x/text v0.18.0
 	k8s.io/api v0.31.0
 	k8s.io/apimachinery v0.31.0
 	k8s.io/cli-runtime v0.31.0
@@ -298,17 +298,17 @@ require (
 	go.uber.org/ratelimit v0.3.0 // indirect
 	go4.org/netipx v0.0.0-20230125063823-8449b0a6169f // indirect
 	golang.org/x/arch v0.4.0 // indirect
-	golang.org/x/crypto v0.25.0 // indirect
+	golang.org/x/crypto v0.27.0 // indirect
 	golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa // indirect
 	golang.org/x/mod v0.19.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/term v0.22.0 // indirect
+	golang.org/x/sys v0.25.0 // indirect
+	golang.org/x/term v0.24.0 // indirect
 	golang.org/x/time v0.6.0 // indirect
 	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240528184218-531527333157 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240701130421-f6361c86f094 // indirect
-	google.golang.org/grpc v1.65.0 // indirect
+	google.golang.org/grpc v1.65.0
 	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
@@ -327,3 +327,5 @@ require (
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 	sigs.k8s.io/yaml v1.4.0 // indirect
 )
+
+retract v1.50.0