test: add tests for s3 fetcher, workerpool and populator

Signed-off-by: Adam Shannag <shannagadam11@gmail.com>

Author: AdamShannag

internal/populator/populator_test.go

@@ -0,0 +1,191 @@
+package populator_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"github.com/AdamShannag/volare/internal/populator"
+	"github.com/AdamShannag/volare/pkg/fetcher"
+	"github.com/AdamShannag/volare/pkg/types"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"strings"
+	"sync"
+	"testing"
+)
+
+type mockFetcher struct {
+	mu       sync.Mutex
+	Called   []types.Source
+	Fail     bool
+	FetchErr error
+}
+
+func (m *mockFetcher) Fetch(_ context.Context, _ string, src types.Source) error {
+	m.mu.Lock()
+	m.Called = append(m.Called, src)
+	m.mu.Unlock()
+
+	if m.Fail {
+		return m.FetchErr
+	}
+	return nil
+}
+
+func TestPopulate_Success(t *testing.T) {
+	t.Parallel()
+
+	reg := fetcher.NewRegistry()
+	mock := &mockFetcher{}
+	_ = reg.Register("s3", mock)
+
+	spec := newValidSpec(t)
+	err := populator.Populate(context.Background(), spec, "/tmp/populate", reg)
+	if err != nil {
+		t.Fatalf("expected success, got: %v", err)
+	}
+	if len(mock.Called) != 2 {
+		t.Errorf("expected 2 fetch calls, got: %d", len(mock.Called))
+	}
+}
+
+func TestPopulate_FetcherGetFails(t *testing.T) {
+	t.Parallel()
+
+	reg := fetcher.NewRegistry()
+
+	spec := newValidSpec(t)
+	err := populator.Populate(context.Background(), spec, "/tmp", reg)
+	if err == nil {
+		t.Fatal("expected fetcher not found error, got nil")
+	}
+}
+
+func TestPopulate_FetchFails(t *testing.T) {
+	t.Parallel()
+
+	reg := fetcher.NewRegistry()
+	mock := &mockFetcher{Fail: true, FetchErr: errors.New("boom")}
+	_ = reg.Register("s3", mock)
+
+	spec := newValidSpec(t)
+	err := populator.Populate(context.Background(), spec, "/tmp", reg)
+	if err == nil {
+		t.Fatal("expected fetch error, got nil")
+	}
+}
+
+func TestPopulate_InvalidSpecJSON(t *testing.T) {
+	t.Parallel()
+
+	err := populator.Populate(context.Background(), `{"invalid":`, "/tmp", nil)
+	if err == nil {
+		t.Fatal("expected JSON error, got nil")
+	}
+}
+
+func TestPopulate_EmptySpec(t *testing.T) {
+	t.Parallel()
+
+	err := populator.Populate(context.Background(), "", "/tmp", nil)
+	if err == nil {
+		t.Fatal("expected error for empty spec, got nil")
+	}
+}
+
+func TestArgsFactory_Success(t *testing.T) {
+	t.Setenv("FOO", "bar")
+
+	vp := types.VolarePopulator{
+		TypeMeta:   metav1.TypeMeta{Kind: "VolarePopulator", APIVersion: "volare/v1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "test-populator"},
+		Spec: types.VolarePopulatorSpec{
+			Sources: []types.Source{
+				{Type: "http", TargetPath: "path/to/target"},
+			},
+			Workers: nil,
+		},
+	}
+
+	unstructuredMap, err := toUnstructured(vp)
+	if err != nil {
+		t.Fatalf("failed to convert to unstructured: %v", err)
+	}
+	u := &unstructured.Unstructured{Object: unstructuredMap}
+
+	mountPath := "/mnt/test"
+	argsFunc := populator.ArgsFactory(mountPath)
+
+	args, err := argsFunc(false, u)
+	if err != nil {
+		t.Fatalf("ArgsFactory returned error: %v", err)
+	}
+
+	if len(args) != 4 {
+		t.Fatalf("expected 4 args, got %d: %v", len(args), args)
+	}
+
+	if !strings.HasPrefix(args[0], "--mode=populator") {
+		t.Errorf("expected args[0] to start with --mode=populator, got %s", args[0])
+	}
+
+	if !strings.HasPrefix(args[1], "--spec=") {
+		t.Errorf("expected args[1] to start with --spec=, got %s", args[1])
+	}
+
+	if !strings.HasPrefix(args[2], "--envs=") {
+		t.Errorf("expected args[2] to start with --envs=, got %s", args[2])
+	}
+
+	if !strings.HasPrefix(args[3], "--mountpath=") {
+		t.Errorf("expected args[3] to start with --mountpath=, got %s", args[3])
+	}
+
+	if args[3] != "--mountpath="+mountPath {
+		t.Errorf("expected mountpath arg %q, got %q", "--mountpath="+mountPath, args[3])
+	}
+}
+
+func TestArgsFactory_InvalidUnstructured(t *testing.T) {
+	t.Parallel()
+
+	u := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"spec": func() {},
+		},
+	}
+
+	argsFunc := populator.ArgsFactory("/mnt/test")
+
+	args, err := argsFunc(false, u)
+	if err == nil {
+		t.Fatal("expected error converting invalid unstructured, got nil")
+	}
+	if len(args) != 0 {
+		t.Errorf("expected empty args slice on error, got %v", args)
+	}
+}
+
+func toUnstructured(obj interface{}) (map[string]interface{}, error) {
+	b, err := json.Marshal(obj)
+	if err != nil {
+		return nil, err
+	}
+	var m map[string]interface{}
+	err = json.Unmarshal(b, &m)
+	return m, err
+}
+
+func newValidSpec(t *testing.T) string {
+	spec := types.VolarePopulatorSpec{
+		Sources: []types.Source{
+			{Type: "s3", TargetPath: "file1.txt"},
+			{Type: "s3", TargetPath: "file2.txt"},
+		},
+	}
+	specBytes, err := json.Marshal(spec)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return string(specBytes)
+}

pkg/fetcher/s3/adapter.go

@@ -0,0 +1,23 @@
+package s3
+
+import (
+	"context"
+	"github.com/minio/minio-go/v7"
+	"io"
+)
+
+type minioAdapter struct {
+	client *minio.Client
+}
+
+func (m *minioAdapter) ListObjects(ctx context.Context, bucket string, opts minio.ListObjectsOptions) <-chan minio.ObjectInfo {
+	return m.client.ListObjects(ctx, bucket, opts)
+}
+
+func (m *minioAdapter) GetObject(ctx context.Context, bucket, object string, opts minio.GetObjectOptions) (io.ReadCloser, error) {
+	obj, err := m.client.GetObject(ctx, bucket, object, opts)
+	if err != nil {
+		return nil, err
+	}
+	return obj, nil
+}

pkg/fetcher/s3/fetcher.go

@@ -19,7 +19,7 @@ import (
 
 type Client interface {
 	ListObjects(ctx context.Context, bucket string, opts minio.ListObjectsOptions) <-chan minio.ObjectInfo
-	GetObject(ctx context.Context, bucket, object string, opts minio.GetObjectOptions) (*minio.Object, error)
+	GetObject(ctx context.Context, bucket, object string, opts minio.GetObjectOptions) (io.ReadCloser, error)
 }
 
 type ClientFactory func(opts types.S3Options) (Client, error)
@@ -137,9 +137,13 @@ func downloadObject(ctx context.Context, client Client, mountPath, bucket, key s
 }
 
 func MinioClientFactory(opts types.S3Options) (Client, error) {
-	return minio.New(opts.Endpoint, &minio.Options{
+	c, err := minio.New(opts.Endpoint, &minio.Options{
 		Creds:  credentials.NewStaticV4(utils.FromEnv(opts.AccessKeyID), utils.FromEnv(opts.SecretAccessKey), utils.FromEnv(opts.SessionToken)),
 		Secure: opts.Secure,
 		Region: opts.Region,
 	})
+	if err != nil {
+		return nil, err
+	}
+	return &minioAdapter{client: c}, nil
 }

pkg/fetcher/s3/fetcher_test.go

@@ -0,0 +1,146 @@
+package s3_test
+
+import (
+	"context"
+	"errors"
+	"io"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/AdamShannag/volare/pkg/fetcher/s3"
+	"github.com/AdamShannag/volare/pkg/types"
+	"github.com/minio/minio-go/v7"
+)
+
+type mockClient struct {
+	listObjectsFunc func(ctx context.Context, bucket string, opts minio.ListObjectsOptions) <-chan minio.ObjectInfo
+	getObjectFunc   func(ctx context.Context, bucket, object string, opts minio.GetObjectOptions) (io.ReadCloser, error)
+}
+
+func (m *mockClient) ListObjects(ctx context.Context, bucket string, opts minio.ListObjectsOptions) <-chan minio.ObjectInfo {
+	return m.listObjectsFunc(ctx, bucket, opts)
+}
+
+func (m *mockClient) GetObject(ctx context.Context, bucket, object string, opts minio.GetObjectOptions) (io.ReadCloser, error) {
+	return m.getObjectFunc(ctx, bucket, object, opts)
+}
+
+func TestFetcher_Fetch_Success(t *testing.T) {
+	t.Parallel()
+
+	objects := []minio.ObjectInfo{
+		{Key: "file1.txt"},
+		{Key: "file2.txt"},
+		{Key: "dir/"},
+	}
+
+	var calls int32
+	mock := &mockClient{
+		listObjectsFunc: func(_ context.Context, _ string, _ minio.ListObjectsOptions) <-chan minio.ObjectInfo {
+			ch := make(chan minio.ObjectInfo, len(objects))
+			for _, o := range objects {
+				ch <- o
+			}
+			close(ch)
+			return ch
+		},
+		getObjectFunc: func(_ context.Context, _, _ string, _ minio.GetObjectOptions) (io.ReadCloser, error) {
+			atomic.AddInt32(&calls, 1)
+			return io.NopCloser(strings.NewReader("data")), nil
+		},
+	}
+
+	fetcher := s3.NewFetcher(func(opts types.S3Options) (s3.Client, error) {
+		return mock, nil
+	})
+
+	src := types.Source{
+		S3: &types.S3Options{
+			Bucket: "bucket",
+			Paths:  []string{""},
+		},
+	}
+
+	tmpDir := t.TempDir()
+	err := fetcher.Fetch(context.Background(), tmpDir, src)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if got := atomic.LoadInt32(&calls); got != 2 {
+		t.Errorf("expected 2 downloads, got %d", got)
+	}
+}
+
+func TestFetcher_Fetch_ListObjectsError(t *testing.T) {
+	t.Parallel()
+
+	mock := &mockClient{
+		listObjectsFunc: func(_ context.Context, _ string, _ minio.ListObjectsOptions) <-chan minio.ObjectInfo {
+			ch := make(chan minio.ObjectInfo, 1)
+			ch <- minio.ObjectInfo{Err: errors.New("list error")}
+			close(ch)
+			return ch
+		},
+		getObjectFunc: func(_ context.Context, _, _ string, _ minio.GetObjectOptions) (io.ReadCloser, error) {
+			t.Fatal("GetObject should not be called")
+			return nil, nil
+		},
+	}
+
+	fetcher := s3.NewFetcher(func(opts types.S3Options) (s3.Client, error) {
+		return mock, nil
+	})
+
+	err := fetcher.Fetch(context.Background(), t.TempDir(), types.Source{
+		S3: &types.S3Options{
+			Bucket: "bucket",
+			Paths:  []string{"/bad"},
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "failed to list objects") {
+		t.Fatalf("expected list error, got %v", err)
+	}
+}
+
+func TestFetcher_Fetch_DownloadError(t *testing.T) {
+	t.Parallel()
+
+	mock := &mockClient{
+		listObjectsFunc: func(_ context.Context, _ string, _ minio.ListObjectsOptions) <-chan minio.ObjectInfo {
+			ch := make(chan minio.ObjectInfo, 1)
+			ch <- minio.ObjectInfo{Key: "file.txt"}
+			close(ch)
+			return ch
+		},
+		getObjectFunc: func(_ context.Context, _, _ string, _ minio.GetObjectOptions) (io.ReadCloser, error) {
+			return nil, errors.New("download error")
+		},
+	}
+
+	fetcher := s3.NewFetcher(func(opts types.S3Options) (s3.Client, error) {
+		return mock, nil
+	})
+
+	err := fetcher.Fetch(context.Background(), t.TempDir(), types.Source{
+		S3: &types.S3Options{
+			Bucket: "bucket",
+			Paths:  []string{"a"},
+		},
+	})
+	if err == nil || !strings.Contains(err.Error(), "download error") {
+		t.Fatalf("expected download error, got %v", err)
+	}
+}
+
+func TestFetcher_Fetch_InvalidConfig(t *testing.T) {
+	t.Parallel()
+
+	fetcher := s3.NewFetcher(nil)
+
+	err := fetcher.Fetch(context.Background(), t.TempDir(), types.Source{})
+	if err == nil || !strings.Contains(err.Error(), "invalid source configuration") {
+		t.Fatalf("expected invalid source config error, got %v", err)
+	}
+}