test: add test for server /v1alpha1/replicas POST GET DELETE

server/http/server_test.go

+package http
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	nervexv1alpha1 "go-sensephoenix.sensetime.com/nervex-operator/api/v1alpha1"
+	servertypes "go-sensephoenix.sensetime.com/nervex-operator/server/types"
+	nervexutil "go-sensephoenix.sensetime.com/nervex-operator/utils"
+	testutil "go-sensephoenix.sensetime.com/nervex-operator/utils/testutils"
+)
+
+var _ = Describe("Server Test", func() {
+	Context("When send request to server", func() {
+		It("Should create the right number of replicas", func() {
+			job := testutil.NewNerveXJob()
+			name := nervexutil.GenerateName(job.Name)
+			job.SetName(name)
+
+			By(fmt.Sprintf("Create %dth NerveXJob", 1))
+			var err error
+			ctx := context.Background()
+			err = creatNerveXJob(ctx, job)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Send request on POST /v1alpha1/replicas")
+			coorname := nervexutil.ReplicaPodName(job.Name, "coordinator")
+			addr := fmt.Sprintf("%s:%d", localServingHost, localServingPort)
+			rurl := fmt.Sprintf("http://%s/v1alpha1/replicas", addr)
+			var cn, ln int = 2, 3
+			req := servertypes.NerveXJobRequest{
+				Namespace:   job.Namespace,
+				Coordinator: coorname,
+				Collectors: servertypes.ResourceQuantity{
+					Replicas: cn,
+				},
+				Learners: servertypes.ResourceQuantity{
+					Replicas: ln,
+				},
+			}
+
+			njresp, err := sendRequest(http.MethodPost, req, rurl)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(len(njresp.Collectors)).Should(Equal(cn))
+			Expect(len(njresp.Learners)).Should(Equal(ln))
+
+			By("Send request on GET /v1alpha1/replicas")
+			gurl := fmt.Sprintf("%s?namespace=%s&coordinator=%s", rurl, job.Namespace, coorname)
+			resp, err := http.Get(gurl)
+			Expect(err).NotTo(HaveOccurred())
+			gnjresp, err := parseResponse(resp)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(gnjresp.Collectors)).Should(Equal(cn))
+			Expect(len(gnjresp.Learners)).Should(Equal(ln))
+
+			By("Send request on DELETE /v1alpha1/replicas")
+			var dcn, dln int = 1, 1
+			dreq := servertypes.NerveXJobRequest{
+				Namespace:   job.Namespace,
+				Coordinator: coorname,
+				Collectors: servertypes.ResourceQuantity{
+					Replicas: dcn,
+				},
+				Learners: servertypes.ResourceQuantity{
+					Replicas: dln,
+				},
+			}
+			dnjresp, err := sendRequest(http.MethodDelete, dreq, rurl)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(len(dnjresp.Collectors)).Should(Equal(dcn))
+			Expect(len(dnjresp.Learners)).Should(Equal(dln))
+		})
+	})
+})
+
+func creatNerveXJob(ctx context.Context, job *nervexv1alpha1.NerveXJob) error {
+	var err error
+	err = k8sClient.Create(ctx, job, &client.CreateOptions{})
+	if err != nil {
+		return err
+	}
+
+	By("Create coordinator")
+	ownRefer := metav1.OwnerReference{
+		APIVersion: nervexv1alpha1.GroupVersion.String(),
+		Kind:       nervexv1alpha1.KindNerveXJob,
+		Name:       job.Name,
+		UID:        job.GetUID(),
+		Controller: func(c bool) *bool { return &c }(true),
+	}
+	coorname := nervexutil.ReplicaPodName(job.Name, "coordinator")
+	coorpod := testutil.NewPod(coorname, job.Name, ownRefer)
+	err = k8sClient.Create(ctx, coorpod, &client.CreateOptions{})
+	if err != nil {
+		return err
+	}
+
+	By("Waiting for server's cache to be synced")
+	time.Sleep(250 * time.Millisecond)
+	return nil
+}
+
+func sendRequest(method string, req servertypes.NerveXJobRequest, url string) (*servertypes.NerveXJobResponse, error) {
+	rbody, err := json.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+
+	// Create client
+	reqs, err := http.NewRequest(method, url, bytes.NewReader(rbody))
+	if err != nil {
+		return nil, err
+	}
+	reqs.Header.Add("Content-Type", "application/json")
+
+	// Fetch Request
+	resp, err := http.DefaultClient.Do(reqs)
+	if err != nil {
+		return nil, err
+	}
+
+	return parseResponse(resp)
+}
+
+func parseResponse(resp *http.Response) (*servertypes.NerveXJobResponse, error) {
+	defer resp.Body.Close()
+	var nresp servertypes.Response
+	err := json.NewDecoder(resp.Body).Decode(&nresp)
+	if err != nil {
+		return nil, err
+	}
+
+	Expect(nresp.Success).Should(BeTrue())
+
+	var njresp servertypes.NerveXJobResponse
+	jsonBytes, err := json.Marshal(nresp.Data)
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(jsonBytes, &njresp)
+	if err != nil {
+		return nil, err
+	}
+	return &njresp, nil
+}

server/http/suite_test.go

+/*
+Copyright 2021 The SensePhoenix 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 http
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"net"
+	"path/filepath"
+	"testing"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+
+	nervexv1alpha1 "go-sensephoenix.sensetime.com/nervex-operator/api/v1alpha1"
+	serverdynamic "go-sensephoenix.sensetime.com/nervex-operator/server/dynamic"
+	testutil "go-sensephoenix.sensetime.com/nervex-operator/utils/testutils"
+	//+kubebuilder:scaffold:imports
+)
+
+// These tests use Ginkgo (BDD-style Go testing framework). Refer to
+// http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
+
+const (
+	timeout  = 5 * time.Second
+	interval = 250 * time.Millisecond
+	duration = 500 * time.Millisecond
+
+	localServingHost = "localhost"
+	localServingPort = 8080
+)
+
+// var cfg *rest.Config
+var k8sClient client.Client
+var testEnv *envtest.Environment
+var kubeClient *kubernetes.Clientset
+
+func TestServer(t *testing.T) {
+	RegisterFailHandler(Fail)
+
+	RunSpecsWithDefaultAndCustomReporters(t,
+		"Server Suite",
+		[]Reporter{printer.NewlineReporter{}})
+}
+
+var _ = BeforeSuite(func() {
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
+
+	By("bootstrapping test environment")
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths:     []string{filepath.Join("..", "..", "config", "crd", "bases")},
+		ErrorIfCRDPathMissing: true,
+	}
+
+	cfg, err := testEnv.Start()
+	Expect(err).NotTo(HaveOccurred())
+	Expect(cfg).NotTo(BeNil())
+
+	err = nervexv1alpha1.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
+
+	//+kubebuilder:scaffold:scheme
+
+	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
+	Expect(err).NotTo(HaveOccurred())
+	Expect(k8sClient).NotTo(BeNil())
+
+	By("Apply AggregatorConfig")
+	ctx := context.Background()
+	agconfig := testutil.NewAggregatorConfig()
+
+	err = k8sClient.Create(ctx, testutil.NewNamespace(agconfig.Namespace), &client.CreateOptions{})
+	Expect(err).NotTo(HaveOccurred())
+
+	err = k8sClient.Create(ctx, agconfig, &client.CreateOptions{})
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Agconfig successfully created")
+	key := types.NamespacedName{Namespace: agconfig.Namespace, Name: agconfig.Name}
+	createdAg := nervexv1alpha1.AggregatorConfig{}
+	Eventually(func() bool {
+		err := k8sClient.Get(ctx, key, &createdAg)
+		if err != nil {
+			return false
+		}
+		return true
+	}, timeout, interval).Should(BeTrue())
+
+	kubeClient = kubernetes.NewForConfigOrDie(cfg)
+	dynamicClient := dynamic.NewForConfigOrDie(cfg)
+
+	dif := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, serverdynamic.ResyncPeriod, corev1.NamespaceAll, nil)
+
+	dyi := serverdynamic.NewDynamicInformer(dif)
+
+	// start dynamic informer
+	stopCh := make(chan struct{})
+	go dif.Start(stopCh)
+
+	opts := zap.Options{
+		Development: true,
+	}
+	opts.BindFlags(flag.CommandLine)
+	flag.Parse()
+
+	logger := zap.New(zap.UseFlagOptions(&opts))
+
+	nervexServer := NewNerveXServer(kubeClient, dynamicClient, logger, dyi)
+
+	addrPort := fmt.Sprintf("%s:%d", localServingHost, localServingPort)
+	go func() {
+		err := nervexServer.Start(addrPort)
+		Expect(err).NotTo(HaveOccurred())
+	}()
+
+	// wait for the server to get ready
+	tcpAddr, err := net.ResolveTCPAddr("tcp", addrPort)
+	Expect(err).NotTo(HaveOccurred())
+
+	Eventually(func() error {
+		conn, err := net.DialTCP("tcp", nil, tcpAddr)
+		if err != nil {
+			return err
+		}
+		conn.Close()
+		return nil
+	}, timeout, interval).Should(Succeed())
+}, 60)
+
+var _ = AfterSuite(func() {
+	By("tearing down the test environment")
+	err := testEnv.Stop()
+	Expect(err).NotTo(HaveOccurred())
+})