Merge pull request #3335 from sttts/sttts-sdk-testing-external

🌱 Make exposed testing package work outside

Author: kcp-ci-bot

sdk/testing/config.go

@@ -17,6 +17,8 @@ limitations under the License.
 package testing
 
 import (
+	"sync"
+
 	kcptestingserver "github.com/kcp-dev/kcp/sdk/testing/server"
 )
 
@@ -32,6 +34,9 @@ var (
 		kubeconfigPath       string
 		shardKubeconfigPaths map[string]string
 	}{}
+
+	externalSetupOnce sync.Once
+	externalSetupFn   func() (kubeconfigPath string, shardKubeconfigPaths map[string]string)
 )
 
 // InitSharedKcpServer initializes a shared kcp server fixture. It must be
@@ -45,7 +50,14 @@ func InitSharedKcpServer(opts ...kcptestingserver.Option) {
 // InitExternalServer configures a potentially pre-existing shared external kcp
 // server. It must be called before SharedKcpServer is called. The shard
 // kubeconfigs are optional, but the kubeconfigPath must be provided.
-func InitExternalServer(kubeconfigPath string, shardKubeconfigPaths map[string]string) {
-	externalConfig.kubeconfigPath = kubeconfigPath
-	externalConfig.shardKubeconfigPaths = shardKubeconfigPaths
+func InitExternalServer(fn func() (kubeconfigPath string, shardKubeconfigPaths map[string]string)) {
+	externalSetupFn = fn
+}
+
+func setupExternal() {
+	externalSetupOnce.Do(func() {
+		if externalSetupFn != nil {
+			externalConfig.kubeconfigPath, externalConfig.shardKubeconfigPaths = externalSetupFn()
+		}
+	})
 }

sdk/testing/doc.go

@@ -0,0 +1,19 @@
+/*
+Copyright 2025 The KCP Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package testing provides utilities for testing of and against KCP. This code
+// to be used outside of the KCP repository is experimental and subject to change.
+package testing

sdk/testing/kcp.go

@@ -41,7 +41,7 @@ func PrivateKcpServer(t *testing.T, options ...kcptestingserver.Option) kcptesti
 
 	cfg := &kcptestingserver.Config{Name: serverName}
 	for _, opt := range options {
-		cfg = opt(cfg)
+		opt(cfg)
 	}
 
 	auditPolicyArg := false
@@ -74,18 +74,23 @@ func PrivateKcpServer(t *testing.T, options ...kcptestingserver.Option) kcptesti
 func SharedKcpServer(t *testing.T) kcptestingserver.RunningServer {
 	t.Helper()
 
+	setupExternal()
 	if len(externalConfig.kubeconfigPath) > 0 {
 		// Use a pre-existing external server
 
-		t.Logf("shared kcp server will target configuration %q", externalConfig.kubeconfigPath)
+		t.Logf("Shared kcp server will target configuration %q", externalConfig.kubeconfigPath)
 		s, err := kcptestingserver.NewExternalKCPServer(sharedConfig.Name, externalConfig.kubeconfigPath, externalConfig.shardKubeconfigPaths, filepath.Join(kcptestinghelpers.RepositoryDir(), ".kcp"))
 		require.NoError(t, err, "failed to create persistent server fixture")
 
 		ctx, cancel := context.WithCancel(context.Background())
 		t.Cleanup(cancel)
 
-		err = kcptestingserver.WaitForReady(ctx, t, s.RootShardSystemMasterBaseConfig(t), true)
-		require.NoError(t, err, "error waiting for readiness")
+		rootCfg := s.RootShardSystemMasterBaseConfig(t)
+		t.Logf("Waiting for readiness for server at %s", rootCfg.Host)
+		err = kcptestingserver.WaitForReady(ctx, rootCfg)
+		require.NoError(t, err, "external server is not ready")
+
+		kcptestingserver.MonitorEndpoints(t, rootCfg, "/livez", "/readyz")
 
 		return s
 	}

sdk/testing/server/config.go

@@ -31,29 +31,43 @@ type Config struct {
 }
 
 // Option a function that wish to modify a given kcp configuration.
-type Option func(*Config) *Config
+type Option func(*Config)
 
 // WithScratchDirectories adds custom scratch directories to a kcp configuration.
 func WithScratchDirectories(artifactDir, dataDir string) Option {
-	return func(cfg *Config) *Config {
+	return func(cfg *Config) {
 		cfg.ArtifactDir = artifactDir
 		cfg.DataDir = dataDir
-		return cfg
 	}
 }
 
 // WithCustomArguments applies provided arguments to a given kcp configuration.
 func WithCustomArguments(args ...string) Option {
-	return func(cfg *Config) *Config {
+	return func(cfg *Config) {
 		cfg.Args = args
-		return cfg
 	}
 }
 
-// WithClientCADir sets the client CA directory for a given kcp configuration.
-func WithClientCADir(clientCADir string) Option {
-	return func(cfg *Config) *Config {
+// WithClientCA sets the client CA directory for a given kcp configuration.
+// A client CA will automatically created and the --client-ca configured.
+func WithClientCA(clientCADir string) Option {
+	return func(cfg *Config) {
 		cfg.ClientCADir = clientCADir
-		return cfg
+	}
+}
+
+// WithRunInProcess sets the kcp server to run in process. This requires extra
+// setup of the RunInProcessFunc variable and will only work inside of the kcp
+// repository.
+func WithRunInProcess() Option {
+	return func(cfg *Config) {
+		cfg.RunInProcess = true
+	}
+}
+
+// WithLogToConsole sets the kcp server to log to console.
+func WithLogToConsole() Option {
+	return func(cfg *Config) {
+		cfg.LogToConsole = true
 	}
 }

sdk/testing/server/fixture.go

@@ -36,6 +36,7 @@ import (
 
 	"github.com/egymgmbh/go-prefix-writer/prefixer"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
 	"sigs.k8s.io/yaml"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -91,8 +92,9 @@ func NewFixture(t *testing.T, cfgs ...Config) Fixture {
 	// Launch kcp servers and ensure they are ready before starting the test
 	start := time.Now()
 	t.Log("Starting kcp servers...")
-	wg := sync.WaitGroup{}
-	wg.Add(len(servers))
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	g, ctx := errgroup.WithContext(ctx)
 	for i, srv := range servers {
 		var opts []RunOption
 		if env.LogToConsoleEnvSet() || cfgs[i].LogToConsole {
@@ -105,17 +107,27 @@ func NewFixture(t *testing.T, cfgs ...Config) Fixture {
 		require.NoError(t, err)
 
 		// Wait for the server to become ready
-		go func(s *kcpServer, i int) {
-			defer wg.Done()
+		g.Go(func() error {
+			if err := srv.loadCfg(); err != nil {
+				return err
+			}
+
+			rootCfg := srv.RootShardSystemMasterBaseConfig(t)
+			t.Logf("Waiting for readiness for server at %s", rootCfg.Host)
+			if err := WaitForReady(srv.ctx, rootCfg); err != nil {
+				return err
+			}
 
-			err := s.loadCfg()
-			require.NoError(t, err, "error loading config")
+			if !cfgs[i].RunInProcess {
+				rootCfg := srv.RootShardSystemMasterBaseConfig(t)
+				MonitorEndpoints(t, rootCfg, "/livez", "/readyz")
+			}
 
-			err = WaitForReady(s.ctx, t, s.RootShardSystemMasterBaseConfig(t), !cfgs[i].RunInProcess)
-			require.NoError(t, err, "kcp server %s never became ready: %v", s.name, err)
-		}(srv, i)
+			return nil
+		})
 	}
-	wg.Wait()
+	err := g.Wait()
+	require.NoError(t, err, "failed to start kcp servers")
 
 	for _, s := range servers {
 		scrapeMetricsForServer(t, s)
@@ -250,6 +262,8 @@ func Command(executableName, identity string) []string {
 	binary := executableName
 	if binDir := os.Getenv(kcpBinariesDirEnvDir); binDir == "" && inKcp {
 		binary = filepath.Join(kcptestinghelpers.RepositoryBinDir(), executableName)
+	} else if binDir != "" {
+		binary = filepath.Join(binDir, executableName)
 	}
 
 	if env.NoGoRunEnvSet() || !inKcp {
@@ -290,9 +304,6 @@ func (c *kcpServer) Run(opts ...RunOption) error {
 	})
 	c.ctx = ctx
 
-	commandLine := append(StartKcpCommand("KCP"), c.args...)
-	c.t.Logf("running: %v", strings.Join(commandLine, " "))
-
 	// run kcp start in-process for easier debugging
 	if runOpts.runInProcess {
 		if RunInProcessFunc == nil {
@@ -314,6 +325,9 @@ func (c *kcpServer) Run(opts ...RunOption) error {
 		return nil
 	}
 
+	commandLine := append(StartKcpCommand("KCP"), c.args...)
+	c.t.Logf("running: %v", strings.Join(commandLine, " "))
+
 	// NOTE: do not use exec.CommandContext here. That method issues a SIGKILL when the context is done, and we
 	// want to issue SIGTERM instead, to give the server a chance to shut down cleanly.
 	cmd := exec.Command(commandLine[0], commandLine[1:]...)
@@ -354,7 +368,10 @@ func (c *kcpServer) Run(opts ...RunOption) error {
 
 	if err := cmd.Start(); err != nil {
 		cleanup()
-		return err
+		if os.Getenv(kcpBinariesDirEnvDir) == "" && commandLine[0] == "kcp" {
+			c.t.Log("Consider setting KCP_BINARIES_DIR pointing to a directory with a kcp binary.")
+		}
+		return fmt.Errorf("failed to start kcp: %w", err)
 	}
 
 	c.t.Cleanup(func() {

sdk/testing/server/ready.go

@@ -21,7 +21,6 @@ import (
 	"errors"
 	"fmt"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -31,9 +30,8 @@ import (
 	"k8s.io/client-go/rest"
 )
 
-func WaitForReady(ctx context.Context, t *testing.T, cfg *rest.Config, keepMonitoring bool) error {
-	t.Logf("waiting for readiness for server at %s", cfg.Host)
-
+// WaitForReady waits for /livez and then /readyz to return success.
+func WaitForReady(ctx context.Context, cfg *rest.Config) error {
 	cfg = rest.CopyConfig(cfg)
 	if cfg.NegotiatedSerializer == nil {
 		cfg.NegotiatedSerializer = kubernetesscheme.Codecs.WithoutConversion()
@@ -44,45 +42,53 @@ func WaitForReady(ctx context.Context, t *testing.T, cfg *rest.Config, keepMonit
 		return fmt.Errorf("failed to create unversioned client: %w", err)
 	}
 
-	wg := sync.WaitGroup{}
-	wg.Add(2)
-	for _, endpoint := range []string{"/livez", "/readyz"} {
-		go func(endpoint string) {
-			defer wg.Done()
-			waitForEndpoint(ctx, t, client, endpoint)
-		}(endpoint)
+	if err := waitForEndpoint(ctx, client, "/livez"); err != nil {
+		return fmt.Errorf("server at %s didn't become ready: %w", cfg.Host, err)
 	}
-	wg.Wait()
-	t.Logf("server at %s is ready", cfg.Host)
-
-	if keepMonitoring {
-		for _, endpoint := range []string{"/livez", "/readyz"} {
-			go func(endpoint string) {
-				monitorEndpoint(ctx, t, client, endpoint)
-			}(endpoint)
-		}
+	if err := waitForEndpoint(ctx, client, "/readyz"); err != nil {
+		return fmt.Errorf("server at %s didn't become ready: %w", cfg.Host, err)
 	}
+
 	return nil
 }
 
-func waitForEndpoint(ctx context.Context, t *testing.T, client *rest.RESTClient, endpoint string) {
+func waitForEndpoint(ctx context.Context, client *rest.RESTClient, endpoint string) error {
 	var lastError error
 	if err := wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, time.Minute, true, func(ctx context.Context) (bool, error) {
 		req := rest.NewRequest(client).RequestURI(endpoint)
-		_, err := req.Do(ctx).Raw()
-		if err != nil {
+		if _, err := req.Do(ctx).Raw(); err != nil {
 			lastError = fmt.Errorf("error contacting %s: failed components: %v", req.URL(), unreadyComponentsFromError(err))
 			return false, nil
 		}
-
-		t.Logf("success contacting %s", req.URL())
 		return true, nil
 	}); err != nil && lastError != nil {
-		t.Error(lastError)
+		return lastError
+	}
+	return nil
+}
+
+// MonitorEndpoints keeps watching the given endpoints and fails t on error.
+func MonitorEndpoints(t *testing.T, client *rest.Config, endpoints ...string) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	for _, endpoint := range endpoints {
+		go func(endpoint string) {
+			monitorEndpoint(ctx, t, client, endpoint)
+		}(endpoint)
 	}
 }
 
-func monitorEndpoint(ctx context.Context, t *testing.T, client *rest.RESTClient, endpoint string) {
+func monitorEndpoint(ctx context.Context, t *testing.T, cfg *rest.Config, endpoint string) {
+	cfg = rest.CopyConfig(cfg)
+	if cfg.NegotiatedSerializer == nil {
+		cfg.NegotiatedSerializer = kubernetesscheme.Codecs.WithoutConversion()
+	}
+	client, err := rest.UnversionedRESTClientFor(cfg)
+	if err != nil {
+		t.Errorf("failed to create unversioned client: %v", err)
+		return
+	}
+
 	// we need a shorter deadline than the server, or else:
 	// timeout.go:135] post-timeout activity - time-elapsed: 23.784917ms, GET "/livez" result: Header called after Handler finished
 	if deadline, ok := t.Deadline(); ok {

test/e2e/framework/flags.go

@@ -20,7 +20,6 @@ import (
 	"errors"
 	"flag"
 	"path/filepath"
-	"testing"
 
 	cliflag "k8s.io/component-base/cli/flag"
 	"k8s.io/klog/v2"
@@ -30,13 +29,6 @@ import (
 	kcptestinghelpers "github.com/kcp-dev/kcp/sdk/testing/helpers"
 )
 
-func init() {
-	klog.InitFlags(flag.CommandLine)
-	if err := flag.Lookup("v").Value.Set("4"); err != nil {
-		panic(err)
-	}
-}
-
 var testConfig = struct {
 	kcpKubeconfig       string
 	shardKubeconfigs    map[string]string
@@ -57,15 +49,18 @@ func complete() {
 }
 
 func init() {
+	klog.InitFlags(flag.CommandLine)
+	if err := flag.Lookup("v").Value.Set("4"); err != nil {
+		panic(err)
+	}
+
 	flag.StringVar(&testConfig.kcpKubeconfig, "kcp-kubeconfig", "", "Path to the kubeconfig for a kcp server.")
 	flag.Var(cliflag.NewMapStringString(&testConfig.shardKubeconfigs), "shard-kubeconfigs", "Paths to the kubeconfigs for a kcp shard server in the format <shard-name>=<kubeconfig-path>. If unset, kcp-kubeconfig is used.")
 	flag.BoolVar(&testConfig.useDefaultKCPServer, "use-default-kcp-server", false, "Whether to use server configuration from .kcp/admin.kubeconfig.")
 	flag.StringVar(&testConfig.suites, "suites", "control-plane", "A comma-delimited list of suites to run.")
 
-	// Make testing package call flags.Parse()
-	testing.Init()
-
-	complete()
-
-	kcptesting.InitExternalServer(testConfig.kcpKubeconfig, testConfig.shardKubeconfigs)
+	kcptesting.InitExternalServer(func() (string, map[string]string) {
+		complete()
+		return testConfig.kcpKubeconfig, testConfig.shardKubeconfigs
+	})
 }