/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 storage_test

import (
	"context"
	"fmt"
	"io"
	"reflect"
	"testing"
	"time"

	"github.com/apache/dubbo-kubernetes/api/dds"
	dubboapacheorgv1alpha1 "github.com/apache/dubbo-kubernetes/api/resource/v1alpha1"
	dubbocp "github.com/apache/dubbo-kubernetes/pkg/config/app/dubbo-cp"
	"github.com/apache/dubbo-kubernetes/pkg/core/endpoint"
	"github.com/apache/dubbo-kubernetes/pkg/core/kubeclient/client"
	"github.com/apache/dubbo-kubernetes/pkg/core/model"
	"github.com/apache/dubbo-kubernetes/pkg/core/schema/collection"
	"github.com/apache/dubbo-kubernetes/pkg/core/schema/collections"
	"github.com/apache/dubbo-kubernetes/pkg/core/schema/gvk"
	"github.com/apache/dubbo-kubernetes/pkg/dds/kube/crdclient"
	"github.com/apache/dubbo-kubernetes/pkg/dds/storage"
	"github.com/apache/dubbo-kubernetes/test/util/retry"
	"github.com/stretchr/testify/assert"
	"google.golang.org/protobuf/types/known/anypb"
	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/tools/cache"
)

type fakeConnection struct {
	sends        []*dds.ObserveResponse
	recvChan     chan recvResult
	disconnected bool
}

type recvResult struct {
	request *dds.ObserveRequest
	err     error
}

func (f *fakeConnection) Send(targetRule *storage.VersionedRule, cr *storage.ClientStatus, response *dds.ObserveResponse) error {
	cr.LastPushedTime = time.Now().Unix()
	cr.LastPushedVersion = targetRule
	cr.LastPushNonce = response.Nonce
	cr.PushingStatus = storage.Pushing
	f.sends = append(f.sends, response)
	return nil
}

func (f *fakeConnection) Recv() (*dds.ObserveRequest, error) {
	request := <-f.recvChan

	return request.request, request.err
}

func (f *fakeConnection) Disconnect() {
	f.disconnected = true
}

func TestStorage_CloseEOF(t *testing.T) {
	t.Parallel()

	s := storage.NewStorage(&dubbocp.Config{})
	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "test",
	}, fake)

	fake.recvChan <- recvResult{
		request: nil,
		err:     io.EOF,
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(s.Connection) != 0 {
		t.Error("expected storage to be removed")
	}
}

func TestStorage_CloseErr(t *testing.T) {
	t.Parallel()

	s := storage.NewStorage(&dubbocp.Config{})
	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "test",
	}, fake)

	fake.recvChan <- recvResult{
		request: nil,
		err:     fmt.Errorf("test"),
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(s.Connection) != 0 {
		t.Error("expected storage to be removed")
	}
}

func TestStorage_UnknowType(t *testing.T) {
	t.Parallel()

	s := storage.NewStorage(&dubbocp.Config{})
	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "test",
	}, fake)

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "",
			Type:  "test",
		},
		err: nil,
	}

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "",
			Type:  "",
		},
		err: nil,
	}

	conn := s.Connection[0]

	fake.recvChan <- recvResult{
		request: nil,
		err:     io.EOF,
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(conn.TypeListened) != 0 {
		t.Error("expected no type listened")
	}
}

func TestStorage_StartNonEmptyNonce(t *testing.T) {
	t.Parallel()

	s := storage.NewStorage(&dubbocp.Config{})
	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "test",
	}, fake)

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "test",
			Type:  gvk.AuthenticationPolicy,
		},
		err: nil,
	}

	conn := s.Connection[0]
	fake.recvChan <- recvResult{
		request: nil,
		err:     io.EOF,
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(conn.TypeListened) != 0 {
		t.Error("expected no type listened")
	}
}

func TestStorage_Listen(t *testing.T) {
	t.Parallel()

	s := storage.NewStorage(&dubbocp.Config{})
	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "test",
	}, fake)

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "",
			Type:  gvk.AuthorizationPolicy,
		},
		err: nil,
	}

	conn := s.Connection[0]

	fake.recvChan <- recvResult{
		request: nil,
		err:     io.EOF,
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(conn.TypeListened) == 0 {
		t.Error("expected type listened")
	}

	if !conn.TypeListened[gvk.AuthorizationPolicy] {
		t.Error("expected type listened")
	}
}

func makeClient(t *testing.T, schemas collection.Schemas) crdclient.ConfigStoreCache {
	fake := client.NewFakeClient()
	for _, s := range schemas.All() {
		_, err := fake.Ext().ApiextensionsV1().CustomResourceDefinitions().Create(context.TODO(), &v1.CustomResourceDefinition{
			ObjectMeta: metav1.ObjectMeta{
				Name: fmt.Sprintf("%s.%s", s.Resource().Plural(), s.Resource().Group()),
			},
		}, metav1.CreateOptions{})
		if err != nil {
			return nil
		}
	}
	stop := make(chan struct{})
	config, err := crdclient.New(fake, "")
	if err != nil {
		t.Fatal(err)
	}
	go func() {
		err := config.Start(stop)
		if err != nil {
			t.Error(err)
			return
		}
	}()
	_ = fake.Start(stop)
	cache.WaitForCacheSync(stop, config.HasSynced)
	t.Cleanup(func() {
		close(stop)
	})
	return config
}

func TestStorage_PreNotify(t *testing.T) {
	t.Parallel()

	store := makeClient(t, collections.Rule)
	configName := "name"
	configNamespace := "namespace"
	timeout := retry.Timeout(time.Second * 20)
	for _, c := range collections.Rule.All() {
		name := c.Resource().Kind()
		t.Run(name, func(t *testing.T) {
			r := c.Resource()
			configMeta := model.Meta{
				GroupVersionKind: r.GroupVersionKind(),
				Name:             configName,
			}
			if !r.IsClusterScoped() {
				configMeta.Namespace = configNamespace
			}

			pb, err := r.NewInstance()
			if err != nil {
				t.Fatal(err)
			}

			if _, err := store.Create(model.Config{
				Meta: configMeta,
				Spec: pb,
			}); err != nil {
				t.Fatalf("Create(%v) => got %v", name, err)
			}
			// Kubernetes is eventually consistent, so we allow a short time to pass before we get
			retry.UntilSuccessOrFail(t, func() error {
				cfg := store.Get(r.GroupVersionKind(), configName, configMeta.Namespace)
				if cfg == nil || !reflect.DeepEqual(cfg.Meta, configMeta) {
					return fmt.Errorf("get(%v) => got unexpected object %v", name, cfg)
				}
				return nil
			}, timeout)
			s := storage.NewStorage(&dubbocp.Config{})

			handler := crdclient.NewHandler(s, "dubbo-demo", store)
			err = handler.NotifyWithIndex(c)
			if err != nil {
				t.Fatal(err)
			}

			fake := &fakeConnection{
				recvChan: make(chan recvResult, 1),
			}

			s.Connected(&endpoint.Endpoint{
				ID: "test",
			}, fake)

			fake.recvChan <- recvResult{
				request: &dds.ObserveRequest{
					Nonce: "",
					Type:  c.Resource().GroupVersionKind().String(),
				},
				err: nil,
			}

			assert.Eventually(t, func() bool {
				return len(fake.sends) == 1
			}, 10*time.Second, time.Millisecond)

			if fake.sends[0].Type != c.Resource().GroupVersionKind().String() {
				t.Error("expected rule type")
			}

			if fake.sends[0].Nonce == "" {
				t.Error("expected non empty nonce")
			}

			if fake.sends[0].Data == nil {
				t.Error("expected data")
			}

			if fake.sends[0].Revision != 1 {
				t.Error("expected Rev 1")
			}

			fake.recvChan <- recvResult{
				request: &dds.ObserveRequest{
					Nonce: fake.sends[0].Nonce,
					Type:  c.Resource().GroupVersionKind().String(),
				},
				err: nil,
			}

			conn := s.Connection[0]

			assert.Eventually(t, func() bool {
				return conn.ClientRules[c.Resource().GroupVersionKind().String()].PushingStatus == storage.Pushed
			}, 10*time.Second, time.Millisecond)

			fake.recvChan <- recvResult{
				request: nil,
				err:     io.EOF,
			}

			assert.Eventually(t, func() bool {
				return fake.disconnected
			}, 10*time.Second, time.Millisecond)

			if len(conn.TypeListened) == 0 {
				t.Error("expected type listened")
			}

			if !conn.TypeListened[c.Resource().GroupVersionKind().String()] {
				t.Error("expected type listened")
			}
		})
	}
}

func TestStorage_AfterNotify(t *testing.T) {
	t.Parallel()

	store := makeClient(t, collections.Rule)
	configName := "name"
	configNamespace := "namespace"
	timeout := retry.Timeout(time.Second * 20)
	for _, c := range collections.Rule.All() {
		name := c.Resource().Kind()
		t.Run(name, func(t *testing.T) {
			r := c.Resource()
			configMeta := model.Meta{
				GroupVersionKind: r.GroupVersionKind(),
				Name:             configName,
			}
			if !r.IsClusterScoped() {
				configMeta.Namespace = configNamespace
			}

			pb, err := r.NewInstance()
			if err != nil {
				t.Fatal(err)
			}
			if r.GroupVersionKind().String() == gvk.ServiceNameMapping {
				mapping := pb.(*dubboapacheorgv1alpha1.ServiceNameMapping)
				mapping.InterfaceName = "test"
				mapping.ApplicationNames = []string{
					"test-app",
				}
			}
			if _, err := store.Create(model.Config{
				Meta: configMeta,
				Spec: pb,
			}); err != nil {
				t.Fatalf("Create(%v) => got %v", name, err)
			}
			// Kubernetes is eventually consistent, so we allow a short time to pass before we get
			retry.UntilSuccessOrFail(t, func() error {
				cfg := store.Get(r.GroupVersionKind(), configName, configMeta.Namespace)
				if cfg == nil || !reflect.DeepEqual(cfg.Meta, configMeta) {
					return fmt.Errorf("get(%v) => got unexpected object %v", name, cfg)
				}
				return nil
			}, timeout)
			s := storage.NewStorage(&dubbocp.Config{})
			handler := crdclient.NewHandler(s, "dubbo-demo", store)

			fake := &fakeConnection{
				recvChan: make(chan recvResult, 1),
			}

			s.Connected(&endpoint.Endpoint{
				ID: "test",
			}, fake)

			fake.recvChan <- recvResult{
				request: &dds.ObserveRequest{
					Nonce: "",
					Type:  c.Resource().GroupVersionKind().String(),
				},
				err: nil,
			}

			conn := s.Connection[0]

			assert.Eventually(t, func() bool {
				return conn.TypeListened[c.Resource().GroupVersionKind().String()]
			}, 10*time.Second, time.Millisecond)

			err = handler.NotifyWithIndex(c)
			if err != nil {
				t.Fatal(err)
			}

			assert.Eventually(t, func() bool {
				return len(fake.sends) == 1
			}, 10*time.Second, time.Millisecond)

			if fake.sends[0].Type != c.Resource().GroupVersionKind().String() {
				t.Error("expected rule type")
			}

			if fake.sends[0].Nonce == "" {
				t.Error("expected non empty nonce")
			}

			if fake.sends[0].Data == nil {
				t.Error("expected data")
			}

			if fake.sends[0].Revision != 1 {
				t.Error("expected Rev 1")
			}

			fake.recvChan <- recvResult{
				request: &dds.ObserveRequest{
					Nonce: fake.sends[0].Nonce,
					Type:  c.Resource().GroupVersionKind().String(),
				},
				err: nil,
			}

			assert.Eventually(t, func() bool {
				return conn.ClientRules[c.Resource().GroupVersionKind().String()].PushingStatus == storage.Pushed
			}, 10*time.Second, time.Millisecond)

			fake.recvChan <- recvResult{
				request: nil,
				err:     io.EOF,
			}

			assert.Eventually(t, func() bool {
				return fake.disconnected
			}, 10*time.Second, time.Millisecond)

			if len(conn.TypeListened) == 0 {
				t.Error("expected type listened")
			}

			if !conn.TypeListened[c.Resource().GroupVersionKind().String()] {
				t.Error("expected type listened")
			}
		})
	}
}

func TestStore_MissNotify(t *testing.T) {
	t.Parallel()

	store := makeClient(t, collections.Rule)
	configName := "name"
	configNamespace := "namespace"
	collection.NewSchemasBuilder().MustAdd(collections.DubboApacheOrgV1Alpha1TagRoute).Build()
	tag := collections.DubboApacheOrgV1Alpha1TagRoute.Resource()
	collection.NewSchemasBuilder().MustAdd(collections.DubboApacheOrgV1Alpha1ConditionRoute).Build()
	condition := collections.DubboApacheOrgV1Alpha1ConditionRoute.Resource()
	tagconfigMeta := model.Meta{
		GroupVersionKind: tag.GroupVersionKind(),
		Name:             configName,
	}
	conditionConfigMeta := model.Meta{
		GroupVersionKind: condition.GroupVersionKind(),
		Name:             configName,
	}
	if !tag.IsClusterScoped() {
		tagconfigMeta.Namespace = configNamespace
	}

	tagpb, err := tag.NewInstance()
	if err != nil {
		t.Fatal(err)
	}
	conditionpb, err := condition.NewInstance()
	if err != nil {
		t.Fatal(err)
	}

	if _, err := store.Create(model.Config{
		Meta: conditionConfigMeta,
		Spec: conditionpb,
	}); err != nil {
		t.Fatalf("Create(%v) => got %v", condition.Kind(), err)
	}

	if _, err := store.Create(model.Config{
		Meta: tagconfigMeta,
		Spec: tagpb,
	}); err != nil {
		t.Fatalf("Create(%v) => got %v", tag.Kind(), err)
	}

	s := storage.NewStorage(&dubbocp.Config{})
	tagHanlder := crdclient.NewHandler(s, "dubbo-demo", store)
	conditionHandler := crdclient.NewHandler(s, "dubbo-demo", store)

	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "test",
	}, fake)

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "",
			Type:  condition.GroupVersionKind().String(),
		},
		err: nil,
	}

	conn := s.Connection[0]

	assert.Eventually(t, func() bool {
		return conn.TypeListened[condition.GroupVersionKind().String()]
	}, 10*time.Second, time.Millisecond)

	if err := conditionHandler.NotifyWithIndex(collections.DubboApacheOrgV1Alpha1ConditionRoute); err != nil {
		t.Fatal(err)
	}
	if err := tagHanlder.NotifyWithIndex(collections.DubboApacheOrgV1Alpha1TagRoute); err != nil {
		t.Fatal(err)
	}

	assert.Eventually(t, func() bool {
		return len(fake.sends) == 1
	}, 10*time.Second, time.Millisecond)

	if fake.sends[0].Type != condition.GroupVersionKind().String() {
		t.Error("expected rule type")
	}

	if fake.sends[0].Nonce == "" {
		t.Error("expected non empty nonce")
	}

	if fake.sends[0].Data == nil {
		t.Error("expected data")
	}

	if fake.sends[0].Revision != 1 {
		t.Error("expected Rev 1")
	}

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: fake.sends[0].Nonce,
			Type:  condition.GroupVersionKind().String(),
		},
		err: nil,
	}

	assert.Eventually(t, func() bool {
		return conn.ClientRules[condition.GroupVersionKind().String()].PushingStatus == storage.Pushed
	}, 10*time.Second, time.Millisecond)

	fake.recvChan <- recvResult{
		request: nil,
		err:     io.EOF,
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(conn.TypeListened) == 0 {
		t.Error("expected type listened")
	}

	if !conn.TypeListened[condition.GroupVersionKind().String()] {
		t.Error("expected type listened")
	}

	if len(fake.sends) != 1 {
		t.Error("expected 1 send")
	}
}

type fakeOrigin struct {
	hash int
}

func (f *fakeOrigin) Type() string {
	return gvk.TagRoute
}

func (f *fakeOrigin) Revision() int64 {
	return 1
}

func (f *fakeOrigin) Exact(gen map[string]storage.DdsResourceGenerator, endpoint *endpoint.Endpoint) (*storage.VersionedRule, error) {
	return &storage.VersionedRule{
		Type:     gvk.TagRoute,
		Revision: 1,
		Data:     []*anypb.Any{},
	}, nil
}

type errOrigin struct{}

func (e errOrigin) Type() string {
	return gvk.TagRoute
}

func (e errOrigin) Revision() int64 {
	return 1
}

func (e errOrigin) Exact(gen map[string]storage.DdsResourceGenerator, endpoint *endpoint.Endpoint) (*storage.VersionedRule, error) {
	return nil, fmt.Errorf("test")
}

func TestStorage_MulitiNotify(t *testing.T) {
	t.Parallel()

	s := storage.NewStorage(&dubbocp.Config{})
	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "test",
	}, fake)

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "",
			Type:  gvk.TagRoute,
		},
		err: nil,
	}

	conn := s.Connection[0]

	assert.Eventually(t, func() bool {
		return conn.TypeListened[gvk.TagRoute]
	}, 10*time.Second, time.Millisecond)

	// should err
	conn.RawRuleQueue.Add(&errOrigin{})
	conn.RawRuleQueue.Add(&fakeOrigin{
		hash: 1,
	})
	conn.RawRuleQueue.Add(&fakeOrigin{
		hash: 2,
	})
	conn.RawRuleQueue.Add(&fakeOrigin{
		hash: 3,
	})

	assert.Eventually(t, func() bool {
		return len(fake.sends) == 1
	}, 10*time.Second, time.Millisecond)

	if fake.sends[0].Type != gvk.TagRoute {
		t.Error("expected rule type")
	}

	if fake.sends[0].Nonce == "" {
		t.Error("expected non empty nonce")
	}

	if fake.sends[0].Data == nil {
		t.Error("expected data")
	}

	assert.Eventually(t, func() bool {
		return conn.ClientRules[gvk.TagRoute].PushQueued
	}, 10*time.Second, time.Millisecond)

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: fake.sends[0].Nonce,
			Type:  gvk.TagRoute,
		},
		err: nil,
	}
	assert.Eventually(t, func() bool {
		return conn.ClientRules[gvk.TagRoute].PushingStatus == storage.Pushed
	}, 10*time.Second, time.Millisecond)

	assert.Eventually(t, func() bool {
		return conn.RawRuleQueue.Len() == 0
	}, 10*time.Second, time.Millisecond)

	fake.recvChan <- recvResult{
		request: nil,
		err:     io.EOF,
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(conn.TypeListened) == 0 {
		t.Error("expected type listened")
	}

	if !conn.TypeListened[gvk.TagRoute] {
		t.Error("expected type listened")
	}

	if len(fake.sends) != 1 {
		t.Error("expected 1 send")
	}
}

func TestStorage_Exact(t *testing.T) {
	t.Parallel()

	configName := "name"
	configNamespace := "namespace"
	for _, c := range collections.Rule.All() {
		r := c.Resource()
		name := c.Resource().Kind()
		t.Run(name, func(t *testing.T) {
			configMeta := model.Meta{
				Name:             configName,
				Namespace:        configNamespace,
				GroupVersionKind: r.GroupVersionKind(),
			}

			if !r.IsClusterScoped() {
				configMeta.Namespace = configNamespace
			}

			pb, err := r.NewInstance()
			if err != nil {
				t.Fatal(err)
			}

			if r.GroupVersionKind().String() == gvk.TagRoute {
				route := pb.(*dubboapacheorgv1alpha1.TagRoute)
				route.Key = "test-key"
				route.Tags = []*dubboapacheorgv1alpha1.Tag{
					{
						Name: "zyq",
						Addresses: []string{
							"lxy",
						},
					},
				}
			}

			origin := &storage.OriginImpl{
				Gvk: r.GroupVersionKind().String(),
				Rev: 1,
				Data: []model.Config{
					{
						Meta: configMeta,
						Spec: pb,
					},
				},
			}

			gen := map[string]storage.DdsResourceGenerator{}
			gen[gvk.AuthenticationPolicy] = &storage.AuthenticationGenerator{}
			gen[gvk.AuthorizationPolicy] = &storage.AuthorizationGenerator{}
			gen[gvk.ServiceNameMapping] = &storage.ServiceMappingGenerator{}
			gen[gvk.ConditionRoute] = &storage.ConditionRoutesGenerator{}
			gen[gvk.TagRoute] = &storage.TagRoutesGenerator{}
			gen[gvk.DynamicConfig] = &storage.DynamicConfigsGenerator{}
			generated, err := origin.Exact(gen, &endpoint.Endpoint{})
			assert.Nil(t, err)

			assert.NotNil(t, generated)
			assert.Equal(t, generated.Type, r.GroupVersionKind().String())
			assert.Equal(t, generated.Revision, int64(1))
		})
	}
}

func TestStorage_ReturnMisNonce(t *testing.T) {
	t.Parallel()

	store := makeClient(t, collections.Rule)
	configName := "name"
	configNamespace := "namespace"
	collection.NewSchemasBuilder().MustAdd(collections.DubboApacheOrgV1Alpha1TagRoute).Build()
	tag := collections.DubboApacheOrgV1Alpha1TagRoute.Resource()
	tagconfigMeta := model.Meta{
		GroupVersionKind: tag.GroupVersionKind(),
		Name:             configName,
	}

	if !tag.IsClusterScoped() {
		tagconfigMeta.Namespace = configNamespace
	}

	tagpb, err := tag.NewInstance()
	if err != nil {
		t.Fatal(err)
	}

	if _, err := store.Create(model.Config{
		Meta: tagconfigMeta,
		Spec: tagpb,
	}); err != nil {
		t.Fatalf("Create(%v) => got %v", tag.Kind(), err)
	}

	s := storage.NewStorage(&dubbocp.Config{})
	tagHanlder := crdclient.NewHandler(s, "dubbo-system", store)
	err = tagHanlder.NotifyWithIndex(collections.DubboApacheOrgV1Alpha1TagRoute)
	if err != nil {
		t.Fatal(err)
	}
	fake := &fakeConnection{
		recvChan: make(chan recvResult, 1),
	}

	s.Connected(&endpoint.Endpoint{
		ID: "TEST",
	}, fake)

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "",
			Type:  gvk.TagRoute,
		},
		err: nil,
	}

	assert.Eventually(t, func() bool {
		return len(fake.sends) == 1
	}, 10*time.Second, time.Millisecond)

	if fake.sends[0].Type != gvk.TagRoute {
		t.Error("expected rule type")
	}
	if fake.sends[0].Nonce == "" {
		t.Error("expected non empty nonce")
	}

	if fake.sends[0].Data == nil {
		t.Error("expected data")
	}
	if fake.sends[0].Revision != 1 {
		t.Error("expected revision 1")
	}

	fake.recvChan <- recvResult{
		request: &dds.ObserveRequest{
			Nonce: "test",
			Type:  gvk.TagRoute,
		},
		err: nil,
	}

	conn := s.Connection[0]

	fake.recvChan <- recvResult{
		request: nil,
		err:     io.EOF,
	}

	assert.Eventually(t, func() bool {
		return fake.disconnected
	}, 10*time.Second, time.Millisecond)

	if len(conn.TypeListened) == 0 {
		t.Error("expected type listened")
	}

	if !conn.TypeListened[gvk.TagRoute] {
		t.Error("expected type listened")
	}

	if conn.ClientRules[gvk.TagRoute].PushingStatus == storage.Pushed {
		t.Error("expected not pushed")
	}
}