sk8s

package module
v0.1.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 20, 2026 License: Apache-2.0 Imports: 52 Imported by: 0

README

Sk8s

Sk8s (Skates) is a Go testing framework for Kubernetes resources. It spins up an ephemeral cluster (k3s) via Testcontainers and provides a rich API for deploying and asserting on workloads — pods, deployments, Helm charts, CRDs, and more.

Prerequisites

  • Go 1.25+
  • Docker (or a compatible container runtime supported by Testcontainers)

Installation

go get github.com/docker-hardened-images/sk8s

Quick start

package mypackage_test

import (
    "context"
    "testing"
    "time"

    sk8s "github.com/docker-hardened-images/sk8s"
    "github.com/stretchr/testify/require"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func TestMyPod(t *testing.T) {
    ctx, cancel := context.WithDeadline(t.Context(), time.Now().Add(5*time.Minute))
    defer cancel()

    cluster, err := sk8s.GetCluster(t, ctx)
    require.NoError(t, err)

    pod := &corev1.Pod{
        ObjectMeta: metav1.ObjectMeta{
            Name:      "my-test-pod",
            Namespace: "default",
        },
        Spec: corev1.PodSpec{
            Containers: []corev1.Container{
                {
                    Name:    "test",
                    Image:   "busybox:latest",
                    Command: []string{"sleep", "300"},
                },
            },
            RestartPolicy: corev1.RestartPolicyAlways,
        },
    }

    _, err = cluster.Client().CoreV1().Pods("default").Create(ctx, pod, metav1.CreateOptions{})
    require.NoError(t, err)

    err = cluster.WaitForPod(ctx, "default", "my-test-pod", "running")
    require.NoError(t, err)
}

The cluster is automatically cleaned up when the test finishes.

Use HelmInstall to deploy a chart and then assert on the resulting workloads.

func TestMyChart(t *testing.T) {
    ctx, cancel := context.WithDeadline(t.Context(), time.Now().Add(10*time.Minute))
    defer cancel()

    cluster, err := sk8s.GetCluster(t, ctx)
    require.NoError(t, err)

    // Load the image your chart uses so it is available without a registry pull
    // or if the image is behind a private repository.
    err = cluster.LoadImages(ctx, "my-org/my-app:latest")
    require.NoError(t, err)

    // Install from an OCI registry
    source := sk8s.ChartSourceFromOCI("oci://registry.example.com/charts/my-chart", "1.2.3")

    values := map[string]interface{}{
        "replicaCount": 1,
        "ingress": map[string]interface{}{
            "enabled": true,
        },
    }

    err = cluster.HelmInstall(ctx, "my-release", source,
        sk8s.WithNamespace("my-namespace"),
        sk8s.WithInstallValues(),
    )
    require.NoError(t, err)

    err = cluster.WaitForDeployment(ctx, "my-namespace", "my-release-my-chart")
    require.NoError(t, err)
}

Cluster configuration

GetCluster accepts functional options to customise the cluster before it starts.

cluster, err := sk8s.GetCluster(t, ctx,
    sk8s.WithClusterConfig(sk8s.ClusterConfig{
        APIServerArg: []string{"feature-gates=SomeFeature=true"},
        Disable:      []string{"metrics-server"},
    }),
    sk8s.WithLoggingOptions(true, true), // stream cluster warnings and pod logs
)

DefaultConfig disables metrics-server and enables the ImageVolume feature gate.

ClusterConfig fields
Field Description
APIServerArg Extra kube-apiserver flags
KubeletArg Extra kubelet flags
Disable k3s components to disable (e.g. "metrics-server")
FlannelBackEnd Flannel backend override
ClusterCIRD Pod CIDR override
DisableNetworkPolicy Disable network policy enforcement
PauseImage Custom pause image
WithCustomizers passes additional Testcontainers ContainerCustomizer options directly to the k3s container (e.g. extra environment variables or bind mounts).

Reference

See API Reference

Contributing

See CONTRIBUTING.md.

Security

See SECURITY.md.

License

See LICENSE.

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var DefaultConfig = ClusterConfig{
	APIServerArg: []string{
		"feature-gates=ImageVolume=true",
	},
	Disable: []string{
		"metrics-server",
	},
}

Functions

func ChartSourceFromOCI 

func ChartSourceFromOCI(ref string, version string) chartSource

func ChartSourceFromUrl 

func ChartSourceFromUrl(repoUrl string, chartName string) chartSource

func ChartSourceFromUrlAndAppVersion 

func ChartSourceFromUrlAndAppVersion(repoUrl string, chartName string, appVersion string) chartSource

func ChartSourceFromUrlAndVersion 

func ChartSourceFromUrlAndVersion(repoUrl string, chartName string, chartVersion string) chartSource

func ChartSourceFromUrlAndVersionWithFallback 

func ChartSourceFromUrlAndVersionWithFallback(repoUrl string, chartName string, chartVersion string, fallbackVersion string) chartSource

func ConvertTypeToHelmValues 

func ConvertTypeToHelmValues(v interface{}) (map[string]interface{}, error)

ConvertTypeToHelmValues converts a typed struct to map[string]interface{} for Helm

func NewTLSConfig 

func NewTLSConfig(options ...TLSConfigOption) (*tls.Config, error)

Types

type ClusterConfig 

type ClusterConfig struct {
	APIServerArg         []string `yaml:"kube-apiserver-arg,omitempty"`
	KubeletArg           []string `yaml:"kubelet-arg,omitempty"`
	Disable              []string `yaml:"disable"`
	FlannelBackEnd       string   `yaml:"flannel-backend,omitempty"`
	ClusterCIRD          string   `yaml:"cluster-cidr,omitempty"`
	DisableNetworkPolicy bool     `yaml:"disable-network-policy,omitempty"`
	PauseImage           string   `yaml:"pause-image,omitempty"`
}

func (*ClusterConfig) Marshall 

func (c *ClusterConfig) Marshall() ([]byte, error)

type ClusterOptions 

type ClusterOptions struct {
	// contains filtered or unexported fields
}

type CustomizeClusterOption 

type CustomizeClusterOption func(opts *ClusterOptions) error

func WithClusterConfig 

func WithClusterConfig(config ClusterConfig) CustomizeClusterOption

func WithCustomizers 

func WithCustomizers(customize ...tc.ContainerCustomizer) CustomizeClusterOption

func WithLoggingOptions 

func WithLoggingOptions(clusterWarnings bool, podLogs bool) CustomizeClusterOption

func (CustomizeClusterOption) Customize 

func (opt CustomizeClusterOption) Customize(opts *ClusterOptions) error

type CustomizeHelmInstallOption 

type CustomizeHelmInstallOption func(opts *HelmInstallOptions) error

func WithInstallValues 

func WithInstallValues(values map[string]interface{}) CustomizeHelmInstallOption

func WithNamespace 

func WithNamespace(namespace string) CustomizeHelmInstallOption

func WithPostRenderer 

func WithPostRenderer(postRenderer *postrenderer.PostRenderer) CustomizeHelmInstallOption

func WithPreloadedImages 

func WithPreloadedImages(images ...string) CustomizeHelmInstallOption

func (CustomizeHelmInstallOption) Customize 

func (opt CustomizeHelmInstallOption) Customize(opts *HelmInstallOptions) error

type HelmInstallOptions 

type HelmInstallOptions struct {
	Namespace       string
	Values          map[string]interface{}
	PostRenderer    *postrenderer.PostRenderer
	PreloadedImages []string
}

type LogOptions 

type LogOptions struct {
	// contains filtered or unexported fields
}

type TLSConfigOption 

type TLSConfigOption func(options *TLSConfigOptions) error

func WithCAFile 

func WithCAFile(caFile string) TLSConfigOption

func WithCertKeyPairFiles 

func WithCertKeyPairFiles(certFile, keyFile string) TLSConfigOption

func WithInsecureSkipVerify 

func WithInsecureSkipVerify(insecureSkipTLSverify bool) TLSConfigOption

type TLSConfigOptions 

type TLSConfigOptions struct {
	// contains filtered or unexported fields
}

type TestCluster 

type TestCluster struct {
	// contains filtered or unexported fields
}

func GetCluster 

func GetCluster(t *testing.T, ctx context.Context, opts ...CustomizeClusterOption) (*TestCluster, error)

func (*TestCluster) ApiExtClient 

func (c *TestCluster) ApiExtClient(ctx context.Context) (*apiextensionsclientset.Clientset, error)

func (*TestCluster) ApplyLocalYAMLs 

func (c *TestCluster) ApplyLocalYAMLs(ctx context.Context, yamlFiles []string) error

ApplyLocalYAMLs copies local YAML files into the cluster and applies them using kubectl. This is useful for CRDs and custom resources that may not be registered in the Go scheme. Unlike ApplyYAMLFile, this method uses kubectl directly, which handles any resource type.

func (*TestCluster) ApplyRemoteYAMLs 

func (c *TestCluster) ApplyRemoteYAMLs(ctx context.Context, urls []string) error

ApplyRemoteYAMLs downloads and applies yamls from remote URLs using kubectl no need to parse multiple "---" documents as kubectl apply handles them natively

func (*TestCluster) ApplyYAMLData 

func (c *TestCluster) ApplyYAMLData(ctx context.Context, yamlData []byte, fieldManager string, extraSchema ...runtime.SchemeBuilder) error

ApplyYAMLData should act like `kubectl apply -f <file>`, but for a yaml doc already in memory. fieldManager should be a string that identifies the caller, usually "test".

func (*TestCluster) ApplyYAMLFile 

func (c *TestCluster) ApplyYAMLFile(ctx context.Context, yamlFile string, fieldManager string, extraSchema ...runtime.SchemeBuilder) error

ApplyYAMLFile should act like `kubectl apply -f <file>`. fieldManager should be a string that identifies the caller, usually "test".

func (*TestCluster) Client 

func (c *TestCluster) Client() *kubernetes.Clientset

func (*TestCluster) Cluster 

func (c *TestCluster) Cluster() *k3s.K3sContainer

func (*TestCluster) CreateMockCert 

func (c *TestCluster) CreateMockCert(ctx context.Context, namespace string, certName string, caCrt, tlsCret, tlsKey []byte) error

func (*TestCluster) CreateNamespace 

func (c *TestCluster) CreateNamespace(ctx context.Context, namespace string) error

CreateNamespace creates a namespace if it doesn't exist

func (*TestCluster) DebugGoDeployment 

func (c *TestCluster) DebugGoDeployment(t *testing.T, namespace string, deploymentName string, cmd string) error

TODO: make debug image customizable DebugGoDeployment modifies a deployment to image mount the go debugger and replaces the command and argument with the debugging one. Example: dhik8s.DebugGoDeployment(t, cluster, client, "default", "my-deployment", "/usr/local/bin/my-cmd")

func (*TestCluster) DeleteResource 

func (c *TestCluster) DeleteResource(ctx context.Context, gvr schema.GroupVersionResource, namespace, name string) error

DeleteResource deletes any Kubernetes resource by name and type

func (*TestCluster) DumpPodsStatus 

func (c *TestCluster) DumpPodsStatus(ctx context.Context, pods []corev1.Pod)

func (*TestCluster) DynamicClient 

func (c *TestCluster) DynamicClient(ctx context.Context) (dynamic.Interface, error)

func (*TestCluster) Exec 

func (c *TestCluster) Exec(ctx context.Context, cmd []string, options ...exec.ProcessOption) (int, io.Reader, error)

func (*TestCluster) ExecPod 

func (c *TestCluster) ExecPod(ctx context.Context, namespace string, pod string, cmd []string) (bytes.Buffer, bytes.Buffer, error)

func (*TestCluster) ExecPodContainer 

func (c *TestCluster) ExecPodContainer(ctx context.Context, namespace string, pod string, container string, cmd []string) (bytes.Buffer, bytes.Buffer, error)

func (*TestCluster) GetPodsForDaemonSet 

func (c *TestCluster) GetPodsForDaemonSet(ctx context.Context, namespace string, dsName string) ([]corev1.Pod, error)

func (*TestCluster) GetPodsForDeployment 

func (c *TestCluster) GetPodsForDeployment(ctx context.Context, namespace string, deploymentName string) ([]corev1.Pod, error)

GetPodsForDeployment retrieves pods managed by a specific deployment.

func (*TestCluster) GetPodsForStatefulSet 

func (c *TestCluster) GetPodsForStatefulSet(ctx context.Context, namespace string, stsName string) ([]corev1.Pod, error)

func (*TestCluster) GetRaw 

func (c *TestCluster) GetRaw(ctx context.Context, apiPath string) (string, error)

GetRaw performs a raw GET request to the Kubernetes API, equivalent to 'kubectl get --raw <apiPath>'

func (*TestCluster) HelmGetNamespace 

func (c *TestCluster) HelmGetNamespace(ctx context.Context) (string, error)

func (*TestCluster) HelmInstall 

func (c *TestCluster) HelmInstall(ctx context.Context, releaseName string, chartSource chartSource, opts ...CustomizeHelmInstallOption) error

func (*TestCluster) HelmSetNamespace 

func (c *TestCluster) HelmSetNamespace(ctx context.Context, namespace string) error

func (*TestCluster) HelmSettings 

func (c *TestCluster) HelmSettings(ctx context.Context) (*cli.EnvSettings, error)

func (*TestCluster) HelmUninstall 

func (c *TestCluster) HelmUninstall(ctx context.Context, releaseName string) error

func (*TestCluster) LoadImages 

func (c *TestCluster) LoadImages(ctx context.Context, images ...string) error

Use for helm chart tests

func (*TestCluster) LoadImagesWithPlatform 

func (c *TestCluster) LoadImagesWithPlatform(ctx context.Context, images []string, platform *ociv1.Platform) error

func (*TestCluster) RunJob 

func (c *TestCluster) RunJob(ctx context.Context, namespace string, job *batchv1.Job) (string, error)

Execute a job running a pod with a given command

func (*TestCluster) ScaleDeployment added in v0.1.1 

func (c *TestCluster) ScaleDeployment(ctx context.Context, namespace, deploymentName string, replicas int32) error

ScaleDeployment sets a Deployment replica count via the scale subresource.

func (*TestCluster) SetupGoDebug 

func (c *TestCluster) SetupGoDebug(t *testing.T, delveVersion string) error

To use in tests add: 

err = sk8s.SetupClusterGoDebug(t, cluster, "1.25")
require.NoError(t, err)

func (*TestCluster) WaitFor 

func (c *TestCluster) WaitFor(ctx context.Context, fn func(ctx context.Context) (bool, error)) error

func (*TestCluster) WaitForAPIResource 

func (c *TestCluster) WaitForAPIResource(ctx context.Context, apiGroup string, resourceName string, kind string) error

Checks that an API resource exists

func (*TestCluster) WaitForCRD 

func (c *TestCluster) WaitForCRD(ctx context.Context, crdName string) error

Checks that CRD exists and is in Established state

func (*TestCluster) WaitForDaemonSet 

func (c *TestCluster) WaitForDaemonSet(ctx context.Context, namespace string, dsName string) error

func (*TestCluster) WaitForDeployment 

func (c *TestCluster) WaitForDeployment(ctx context.Context, namespace string, deploymentName string) error

func (*TestCluster) WaitForDeploymentReplicaCount added in v0.1.1 

func (c *TestCluster) WaitForDeploymentReplicaCount(ctx context.Context, namespace, deploymentName string, want int32) error

WaitForDeploymentReplicaCount blocks until the Deployment's status reflects the desired replica count. For want > 0 it also requires ReadyReplicas to match. For want == 0 it returns when Replicas is zero (scaled-down), without requiring ready pods.

func (*TestCluster) WaitForEvent 

func (c *TestCluster) WaitForEvent(ctx context.Context, namespace string, gvrName string, resourceKind string, expectedReason string, containedMessage string) error

func (*TestCluster) WaitForExternalService 

func (c *TestCluster) WaitForExternalService(ctx context.Context, namespace string, serviceName string) error

func (*TestCluster) WaitForGVR 

func (c *TestCluster) WaitForGVR(ctx context.Context, namespace string, gvr schema.GroupVersionResource, gvrName string, expectedValue string, fields ...string) error

Wait for a specific attribute value in a GroupVersionResource

func (*TestCluster) WaitForGVRBool 

func (c *TestCluster) WaitForGVRBool(ctx context.Context, namespace string, gvr schema.GroupVersionResource, gvrName string, expectedValue bool, fields ...string) error

Wait for a specific attribute value in a GroupVersionResource

func (*TestCluster) WaitForGVRDeletion 

func (c *TestCluster) WaitForGVRDeletion(ctx context.Context, namespace string, gvr schema.GroupVersionResource, gvrName string) error

Wait for a specific attribute value in a GroupVersionResource

func (*TestCluster) WaitForJob 

func (c *TestCluster) WaitForJob(ctx context.Context, namespace string, jobName string) error

func (*TestCluster) WaitForLog 

func (c *TestCluster) WaitForLog(ctx context.Context, namespace string, pod string, log string) error

func (*TestCluster) WaitForLogLineWithOptions 

func (c *TestCluster) WaitForLogLineWithOptions(ctx context.Context, namespace string, pod string, log []string, options *corev1.PodLogOptions) error

func (*TestCluster) WaitForLogWithOptions 

func (c *TestCluster) WaitForLogWithOptions(ctx context.Context, namespace string, pod string, log string, options *corev1.PodLogOptions) error

func (*TestCluster) WaitForPVC 

func (c *TestCluster) WaitForPVC(ctx context.Context, namespace string, pvcName string) error

func (*TestCluster) WaitForPod 

func (c *TestCluster) WaitForPod(ctx context.Context, namespace string, pod string, state string) error

func (*TestCluster) WaitForPodByLabel 

func (c *TestCluster) WaitForPodByLabel(ctx context.Context, namespace string, labelSelector string) (*corev1.Pod, error)

func (*TestCluster) WaitForSecret 

func (c *TestCluster) WaitForSecret(ctx context.Context, namespace string, secretName string) error

func (*TestCluster) WaitForService 

func (c *TestCluster) WaitForService(ctx context.Context, namespace string, serviceName string) error

func (*TestCluster) WaitForState 

func (c *TestCluster) WaitForState(ctx context.Context, namespace string, pod string, state string) error

func (*TestCluster) WaitForStatefulSet 

func (c *TestCluster) WaitForStatefulSet(ctx context.Context, namespace string, stsName string) error

func (*TestCluster) WaitForWithTimeout 

func (c *TestCluster) WaitForWithTimeout(ctx context.Context, interval time.Duration, timeout time.Duration, immediate bool, condition kwait.ConditionWithContextFunc) error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.