wenshu/k8s-orchestrator/test/api/server_remaining_test.go

package api_test

import (
	"bytes"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"testing"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
	ctrlclientfake "sigs.k8s.io/controller-runtime/pkg/client/fake"

	orcv1 "github.com/cyrus/spark-orchestrator/api/v1alpha1"
	api "github.com/cyrus/spark-orchestrator/internal/api"
	"github.com/cyrus/spark-orchestrator/internal/service"
)

func TestListNodesOK(t *testing.T) {
	node := &orcv1.NodeInventory{
		ObjectMeta: metav1.ObjectMeta{Name: "node-a"},
		Spec: orcv1.NodeInventorySpec{
			Address:      "10.0.0.10",
			SSHSecretRef: "orchestrator-system/ssh-node-a",
			Capacity:     orcv1.NodeCapacity{CPUCores: 8, RamGB: 32, SSDGB: 256},
		},
		Status: orcv1.NodeInventoryStatus{
			Phase:              orcv1.NodePhaseIdle,
			KubernetesNodeName: "k8s-node-a",
		},
	}

	h := newTestServerWithObjects(t, node)
	req := httptest.NewRequest(http.MethodGet, "/v1/nodes", nil)
	resp := httptest.NewRecorder()

	h.ServeHTTP(resp, req)
	if resp.Code != http.StatusOK {
		t.Fatalf("expected 200, got %d", resp.Code)
	}

	var payload struct {
		Items []orcv1.NodeInventory `json:"items"`
	}
	if err := json.Unmarshal(resp.Body.Bytes(), &payload); err != nil {
		t.Fatalf("unmarshal response: %v", err)
	}
	if len(payload.Items) != 1 {
		t.Fatalf("expected 1 node, got %d", len(payload.Items))
	}
	if payload.Items[0].Name != "node-a" {
		t.Fatalf("expected node-a, got %s", payload.Items[0].Name)
	}
}

func TestDeleteNodeAccepted(t *testing.T) {
	h := newTestServer(t)
	req := httptest.NewRequest(http.MethodDelete, "/v1/nodes/node-a", nil)
	req.Header.Set("Idempotency-Key", "delete-node-1")
	resp := httptest.NewRecorder()

	h.ServeHTTP(resp, req)
	if resp.Code != http.StatusAccepted {
		t.Fatalf("expected 202, got %d", resp.Code)
	}

	var payload map[string]string
	_ = json.Unmarshal(resp.Body.Bytes(), &payload)
	if payload["resourceId"] != "node-a" {
		t.Fatalf("expected resourceId=node-a, got %q", payload["resourceId"])
	}
	if payload["operationId"] == "" {
		t.Fatal("operationId should not be empty")
	}
}

func TestCreateClusterAcceptedAndGetOperation(t *testing.T) {
	h := newTestServer(t)
	body := []byte(`{"clusterId":"spark-c1","nodeIds":["node-a"],"sparkConfig":{"k":"v"}}`)

	req := httptest.NewRequest(http.MethodPost, "/v1/spark-clusters", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Idempotency-Key", "create-cluster-1")
	resp := httptest.NewRecorder()
	h.ServeHTTP(resp, req)
	if resp.Code != http.StatusAccepted {
		t.Fatalf("expected 202, got %d", resp.Code)
	}

	var createResp map[string]string
	_ = json.Unmarshal(resp.Body.Bytes(), &createResp)
	opID := createResp["operationId"]
	if opID == "" {
		t.Fatal("operationId should not be empty")
	}

	getReq := httptest.NewRequest(http.MethodGet, "/v1/operations/"+opID, nil)
	getResp := httptest.NewRecorder()
	h.ServeHTTP(getResp, getReq)
	if getResp.Code != http.StatusOK {
		t.Fatalf("expected 200, got %d", getResp.Code)
	}
}

func TestGetClusterOK(t *testing.T) {
	cluster := &orcv1.SparkVirtualCluster{
		ObjectMeta: metav1.ObjectMeta{Name: "spark-c1"},
		Spec: orcv1.SparkVirtualClusterSpec{
			NodeRefs: []string{"node-a"},
		},
		Status: orcv1.SparkVirtualClusterStatus{
			Phase: orcv1.SparkClusterPhaseReady,
		},
	}
	h := newTestServerWithObjects(t, cluster)

	req := httptest.NewRequest(http.MethodGet, "/v1/spark-clusters/spark-c1", nil)
	resp := httptest.NewRecorder()
	h.ServeHTTP(resp, req)
	if resp.Code != http.StatusOK {
		t.Fatalf("expected 200, got %d", resp.Code)
	}
}

func TestReleaseScaleAndSubmitJobAccepted(t *testing.T) {
	h := newTestServer(t)
	testCases := []struct {
		name    string
		method  string
		path    string
		idemKey string
		body    string
	}{
		{
			name:    "release cluster",
			method:  http.MethodPost,
			path:    "/v1/spark-clusters/spark-c1/release",
			idemKey: "release-1",
			body:    `{}`,
		},
		{
			name:    "scale out",
			method:  http.MethodPost,
			path:    "/v1/spark-clusters/spark-c1/scale-out",
			idemKey: "scale-out-1",
			body:    `{"nodeIds":["node-a","node-b"]}`,
		},
		{
			name:    "scale in",
			method:  http.MethodPost,
			path:    "/v1/spark-clusters/spark-c1/scale-in",
			idemKey: "scale-in-1",
			body:    `{"nodeIds":["node-a"]}`,
		},
		{
			name:    "submit spark job",
			method:  http.MethodPost,
			path:    "/v1/spark-jobs",
			idemKey: "submit-job-1",
			body:    `{"clusterId":"spark-c1","sparkApplicationSpec":{"type":"Scala","mode":"cluster"}}`,
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			req := httptest.NewRequest(tc.method, tc.path, bytes.NewReader([]byte(tc.body)))
			req.Header.Set("Content-Type", "application/json")
			req.Header.Set("Idempotency-Key", tc.idemKey)
			resp := httptest.NewRecorder()

			h.ServeHTTP(resp, req)
			if resp.Code != http.StatusAccepted {
				t.Fatalf("expected 202, got %d, body=%s", resp.Code, resp.Body.String())
			}
		})
	}
}

func newTestServerWithObjects(t *testing.T, objs ...ctrlclient.Object) http.Handler {
	t.Helper()
	scheme := runtime.NewScheme()
	utilruntime.Must(corev1.AddToScheme(scheme))
	utilruntime.Must(orcv1.AddToScheme(scheme))
	builder := ctrlclientfake.NewClientBuilder().WithScheme(scheme)
	if len(objs) > 0 {
		builder = builder.WithObjects(objs...)
	}
	fakeClient := builder.Build()
	svc := service.NewOrchestratorService(fakeClient, "orchestrator-system")
	return api.NewServer(svc).Router()
}