test(contrib): add unit test for contrib/registry/k8s (#2166)

* test(contrib): add unit test for contrib/registry/k8s

* test(contrib): add unit test for contrib/registry/k8s

Co-authored-by: rogerogers <rogers@rogerogers.com>

contrib/registry/kubernetes/go.sum

@@ -201,6 +201,7 @@ github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpO
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
+github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=

contrib/registry/kubernetes/registry_test.go

@@ -0,0 +1,186 @@
+package kuberegistry
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/go-kratos/kratos/v2/registry"
+
+	appsv1 "k8s.io/api/apps/v1"
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/util/homedir"
+)
+
+const (
+	namespace  = "default"
+	deployName = "hello-deployment"
+	podName    = "hello"
+)
+
+var deployment = appsv1.Deployment{
+	ObjectMeta: metav1.ObjectMeta{
+		Name: deployName,
+	},
+	Spec: appsv1.DeploymentSpec{
+		Replicas: int32Ptr(1),
+		Selector: &metav1.LabelSelector{
+			MatchLabels: map[string]string{
+				"app": podName,
+			},
+		},
+		Template: apiv1.PodTemplateSpec{
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					"app": podName,
+				},
+			},
+			Spec: apiv1.PodSpec{
+				Containers: []apiv1.Container{
+					{
+						Name:  "nginx",
+						Image: "nginx:alpine",
+						Ports: []apiv1.ContainerPort{
+							{
+								Name:          "http",
+								Protocol:      apiv1.ProtocolTCP,
+								ContainerPort: 80,
+							},
+						},
+						Command: []string{
+							"nginx",
+							"-g",
+							"daemon off;",
+						},
+					},
+				},
+			},
+		},
+	},
+}
+
+func getClientSet() (*kubernetes.Clientset, error) {
+	restConfig, err := rest.InClusterConfig()
+	home := homedir.HomeDir()
+
+	if err != nil {
+		kubeconfig := filepath.Join(home, ".kube", "config")
+		restConfig, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
+		if err != nil {
+			return nil, err
+		}
+	}
+	clientSet, err := kubernetes.NewForConfig(restConfig)
+	if err != nil {
+		return nil, err
+	}
+	return clientSet, nil
+}
+
+func int32Ptr(i int32) *int32 { return &i }
+
+func TestSetEnv(t *testing.T) {
+	os.Setenv("HOSTNAME", podName)
+	if os.Getenv("HOSTNAME") != podName {
+		t.Fatal("error")
+	}
+}
+
+func TestRegistry(t *testing.T) {
+	currentNamespace = "default"
+
+	clientSet, err := getClientSet()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	r := NewRegistry(clientSet)
+	r.Start()
+
+	svrHello := &registry.ServiceInstance{
+		ID:        "1",
+		Name:      "hello",
+		Version:   "v1.0.0",
+		Endpoints: []string{"http://127.0.0.1:80"},
+	}
+	_, err = clientSet.AppsV1().Deployments(namespace).Create(context.Background(), &deployment, metav1.CreateOptions{})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	watch, err := r.Watch(context.Background(), svrHello.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer func() {
+		_ = watch.Stop()
+	}()
+
+	go func() {
+		for {
+			res, err1 := watch.Next()
+			if err1 != nil {
+				return
+			}
+			t.Logf("watch: %d", len(res))
+			for _, r := range res {
+				t.Logf("next: %+v", r)
+			}
+		}
+	}()
+	time.Sleep(time.Second)
+
+	pod, err := clientSet.CoreV1().Pods(namespace).List(context.Background(), metav1.ListOptions{
+		LabelSelector: "app=hello",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(pod.Items) < 1 {
+		t.Fatal("fetch resource error")
+	}
+
+	os.Setenv("HOSTNAME", pod.Items[0].Name)
+
+	// Alway remember delete test resource
+	defer func() {
+		_ = clientSet.AppsV1().Deployments(namespace).Delete(context.Background(), deployName, metav1.DeleteOptions{})
+	}()
+
+	if err = r.Register(context.Background(), svrHello); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(time.Second)
+
+	res, err := r.GetService(context.Background(), svrHello.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(res) != 1 && res[0].Name != svrHello.Name {
+		t.Fatal(err)
+	}
+
+	if err1 := r.Deregister(context.Background(), svrHello); err1 != nil {
+		t.Fatal(err1)
+	}
+
+	time.Sleep(time.Second)
+
+	res, err = r.GetService(context.Background(), svrHello.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(res) != 0 {
+		t.Fatal("not expected empty")
+	}
+}