spark-mcp/internal/storage/cluster_repo_test.go

package storage

import (
	"context"
	"errors"
	"reflect"
	"testing"

	"spark-mcp-go/internal/cluster"
)

func newClusterRepo(t *testing.T) *ClusterRepo {
	t.Helper()
	db, err := Open(":memory:")
	if err != nil {
		t.Fatalf("Open in-memory db: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	return db.Clusters()
}

func testClusterA() *cluster.Cluster {
	return &cluster.Cluster{
		ID:                    "cluster-a",
		Name:                  "Cluster Alpha",
		RMURL:                 "http://rm-a.example.com:8088",
		SHSURL:                "http://shs-a.example.com:18080",
		SparkSubmitExecuteBin: "/usr/bin/spark-submit",
		IsActive:              true,
		AuthType:              cluster.AuthBasic,
		AuthUsername:          "admin-a",
		AuthPassword:          "secret-1",
		SSLVerify:             true,
		SSLCABundle:           "/etc/ssl/ca-bundle-a.pem",
		URLAllowlist:          []string{"*.a.example.com", "http://hist-a.example.com/*"},
		DefaultSubmitArgs:     []string{"--master=yarn", "--deploy-mode=cluster"},
		RateLimitPerMin:       10,
	}
}

func assertClusterEqual(t *testing.T, got, want *cluster.Cluster) {
	t.Helper()

	if got.ID != want.ID {
		t.Errorf("ID: got %q, want %q", got.ID, want.ID)
	}
	if got.Name != want.Name {
		t.Errorf("Name: got %q, want %q", got.Name, want.Name)
	}
	if got.RMURL != want.RMURL {
		t.Errorf("RMURL: got %q, want %q", got.RMURL, want.RMURL)
	}
	if got.SHSURL != want.SHSURL {
		t.Errorf("SHSURL: got %q, want %q", got.SHSURL, want.SHSURL)
	}
	if got.SparkSubmitExecuteBin != want.SparkSubmitExecuteBin {
		t.Errorf("SparkSubmitExecuteBin: got %q, want %q", got.SparkSubmitExecuteBin, want.SparkSubmitExecuteBin)
	}
	if got.IsActive != want.IsActive {
		t.Errorf("IsActive: got %v, want %v", got.IsActive, want.IsActive)
	}
	if got.AuthType != want.AuthType {
		t.Errorf("AuthType: got %q, want %q", got.AuthType, want.AuthType)
	}
	if got.AuthUsername != want.AuthUsername {
		t.Errorf("AuthUsername: got %q, want %q", got.AuthUsername, want.AuthUsername)
	}
	if got.AuthPassword != want.AuthPassword {
		t.Errorf("AuthPassword: got %q, want %q", got.AuthPassword, want.AuthPassword)
	}
	if got.SSLVerify != want.SSLVerify {
		t.Errorf("SSLVerify: got %v, want %v", got.SSLVerify, want.SSLVerify)
	}
	if got.SSLCABundle != want.SSLCABundle {
		t.Errorf("SSLCABundle: got %q, want %q", got.SSLCABundle, want.SSLCABundle)
	}
	if !reflect.DeepEqual(got.URLAllowlist, want.URLAllowlist) {
		t.Errorf("URLAllowlist: got %v, want %v", got.URLAllowlist, want.URLAllowlist)
	}
	if !reflect.DeepEqual(got.DefaultSubmitArgs, want.DefaultSubmitArgs) {
		t.Errorf("DefaultSubmitArgs: got %v, want %v", got.DefaultSubmitArgs, want.DefaultSubmitArgs)
	}
	if got.RateLimitPerMin != want.RateLimitPerMin {
		t.Errorf("RateLimitPerMin: got %d, want %d", got.RateLimitPerMin, want.RateLimitPerMin)
	}

	if !got.CreatedAt.Equal(want.CreatedAt) {
		if got.CreatedAt.Unix() != want.CreatedAt.Unix() {
			t.Errorf("CreatedAt: got %v, want %v", got.CreatedAt, want.CreatedAt)
		}
	}
	if !got.UpdatedAt.Equal(want.UpdatedAt) {
		if got.UpdatedAt.Unix() != want.UpdatedAt.Unix() {
			t.Errorf("UpdatedAt: got %v, want %v", got.UpdatedAt, want.UpdatedAt)
		}
	}
}

func TestClusterRepo_Create(t *testing.T) {
	repo := newClusterRepo(t)
	ctx := context.Background()

	want := testClusterA()
	if err := repo.Create(ctx, want); err != nil {
		t.Fatalf("Create: %v", err)
	}

	got, err := repo.Get(ctx, want.ID)
	if err != nil {
		t.Fatalf("Get after Create: %v", err)
	}
	assertClusterEqual(t, got, want)
}

func TestClusterRepo_Get(t *testing.T) {
	repo := newClusterRepo(t)
	ctx := context.Background()

	want := testClusterA()
	if err := repo.Create(ctx, want); err != nil {
		t.Fatalf("Create: %v", err)
	}

	got, err := repo.Get(ctx, want.ID)
	if err != nil {
		t.Fatalf("Get: %v", err)
	}
	assertClusterEqual(t, got, want)

	if _, err := repo.Get(ctx, "nope"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("Get missing cluster: got %v, want ErrNotFound", err)
	}
}

func TestClusterRepo_List(t *testing.T) {
	repo := newClusterRepo(t)
	ctx := context.Background()

	ids := []string{"cluster-a", "cluster-b", "cluster-c"}
	for _, id := range ids {
		c := testClusterA()
		c.ID = id
		c.Name = id
		if err := repo.Create(ctx, c); err != nil {
			t.Fatalf("Create %s: %v", id, err)
		}
		// Sub-second precision in stored timestamps gives us a stable
		// ORDER BY DESC without artificial delays.
	}

	got, err := repo.List(ctx)
	if err != nil {
		t.Fatalf("List: %v", err)
	}
	if len(got) != len(ids) {
		t.Fatalf("List len: got %d, want %d", len(got), len(ids))
	}

	// List must be ordered by created_at DESC: last created first.
	wantOrder := []string{"cluster-c", "cluster-b", "cluster-a"}
	for i, c := range got {
		if c.ID != wantOrder[i] {
			t.Errorf("List order[%d]: got %q, want %q", i, c.ID, wantOrder[i])
		}
	}
}

func TestClusterRepo_Update(t *testing.T) {
	repo := newClusterRepo(t)
	ctx := context.Background()

	c := testClusterA()
	if err := repo.Create(ctx, c); err != nil {
		t.Fatalf("Create: %v", err)
	}

	stored, err := repo.Get(ctx, c.ID)
	if err != nil {
		t.Fatalf("Get before Update: %v", err)
	}

	stored.Name = "Cluster Alpha Updated"
	stored.RMURL = "http://rm-a-new.example.com:8088"
	stored.IsActive = false
	stored.RateLimitPerMin = 20
	// Intentionally leave AuthPassword empty: must preserve existing password.
	stored.AuthPassword = ""

	if err := repo.Update(ctx, stored); err != nil {
		t.Fatalf("Update: %v", err)
	}

	got, err := repo.Get(ctx, c.ID)
	if err != nil {
		t.Fatalf("Get after Update: %v", err)
	}

	if got.Name != stored.Name {
		t.Errorf("Name: got %q, want %q", got.Name, stored.Name)
	}
	if got.RMURL != stored.RMURL {
		t.Errorf("RMURL: got %q, want %q", got.RMURL, stored.RMURL)
	}
	if got.IsActive != false {
		t.Errorf("IsActive: got %v, want false", got.IsActive)
	}
	if got.RateLimitPerMin != 20 {
		t.Errorf("RateLimitPerMin: got %d, want 20", got.RateLimitPerMin)
	}
	if got.AuthPassword != c.AuthPassword {
		t.Errorf("AuthPassword should be preserved: got %q, want %q", got.AuthPassword, c.AuthPassword)
	}

	// Update a non-existing cluster must return ErrNotFound.
	missing := testClusterA()
	missing.ID = "missing"
	if err := repo.Update(ctx, missing); !errors.Is(err, ErrNotFound) {
		t.Fatalf("Update missing cluster: got %v, want ErrNotFound", err)
	}
}

func TestClusterRepo_UpdatePassword(t *testing.T) {
	repo := newClusterRepo(t)
	ctx := context.Background()

	c := testClusterA()
	c.AuthPassword = "old"
	if err := repo.Create(ctx, c); err != nil {
		t.Fatalf("Create: %v", err)
	}

	stored, err := repo.Get(ctx, c.ID)
	if err != nil {
		t.Fatalf("Get before Update: %v", err)
	}
	if stored.AuthPassword != "old" {
		t.Fatalf("initial password: got %q, want %q", stored.AuthPassword, "old")
	}

	stored.AuthPassword = "new"
	if err := repo.Update(ctx, stored); err != nil {
		t.Fatalf("Update password: %v", err)
	}

	got, err := repo.Get(ctx, c.ID)
	if err != nil {
		t.Fatalf("Get after Update: %v", err)
	}
	if got.AuthPassword != "new" {
		t.Errorf("password: got %q, want %q", got.AuthPassword, "new")
	}
}

func TestClusterRepo_UpdateEmptyPasswordKeepsOldPassword(t *testing.T) {
	repo := newClusterRepo(t)
	ctx := context.Background()

	c := testClusterA()
	c.AuthPassword = "keep-me"
	if err := repo.Create(ctx, c); err != nil {
		t.Fatalf("Create: %v", err)
	}

	stored, err := repo.Get(ctx, c.ID)
	if err != nil {
		t.Fatalf("Get before Update: %v", err)
	}
	stored.AuthPassword = ""
	if err := repo.Update(ctx, stored); err != nil {
		t.Fatalf("Update: %v", err)
	}

	got, err := repo.Get(ctx, c.ID)
	if err != nil {
		t.Fatalf("Get after Update: %v", err)
	}
	if got.AuthPassword != "keep-me" {
		t.Errorf("empty password should keep old value: got %q, want %q", got.AuthPassword, "keep-me")
	}
}

func TestClusterRepo_Delete(t *testing.T) {
	repo := newClusterRepo(t)
	ctx := context.Background()

	c := testClusterA()
	if err := repo.Create(ctx, c); err != nil {
		t.Fatalf("Create: %v", err)
	}

	if err := repo.Delete(ctx, c.ID); err != nil {
		t.Fatalf("Delete existing: %v", err)
	}

	if _, err := repo.Get(ctx, c.ID); !errors.Is(err, ErrNotFound) {
		t.Fatalf("Get after Delete: got %v, want ErrNotFound", err)
	}

	if err := repo.Delete(ctx, c.ID); !errors.Is(err, ErrNotFound) {
		t.Fatalf("Delete again: got %v, want ErrNotFound", err)
	}

	if err := repo.Delete(ctx, "nope"); !errors.Is(err, ErrNotFound) {
		t.Fatalf("Delete missing: got %v, want ErrNotFound", err)
	}
}