package ticdc

import (
	"context"
	"testing"

	. "github.com/onsi/gomega"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/stretchr/testify/require"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	"github.com/tidbcloud/infra-api/pkg/nodegroup"

	v1 "github.com/tidbcloud/infra-api/api/v1"
	gitopsv2 "github.com/tidbcloud/infra-api/pkg/gitops-v2"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	"github.com/tidbcloud/infra-api/pkg/provider"
)

func TestDeleteTicdcPvc(t *testing.T) {
	g := NewGomegaWithT(t)
	var objs []client.Object = []client.Object{
		// a match pvc
		&corev1.PersistentVolumeClaim{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: "ns1",
				Name:      "test1",
			},
			Spec: corev1.PersistentVolumeClaimSpec{
				StorageClassName: pointer.String(k8s.TiCDCGp3StorageClassName),
			},
		},
		// namespace not match
		&corev1.PersistentVolumeClaim{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: "ns2",
				Name:      "test2",
			},
			Spec: corev1.PersistentVolumeClaimSpec{
				StorageClassName: pointer.String(k8s.TiCDCGp3StorageClassName),
			},
		},
		// storageclass not match
		&corev1.PersistentVolumeClaim{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: "ns1",
				Name:      "test3",
			},
			Spec: corev1.PersistentVolumeClaimSpec{
				StorageClassName: pointer.String("Unknown"),
			},
		},
	}

	client := fake.NewClientBuilder().WithObjects(objs...).Build()
	ctx := context.Background()

	ctrl := new(ticdcCluster)
	ctrl.destClusterCli = client
	err := ctrl.deleteTicdcPvc(ctx, "ns1")
	g.Expect(err).ShouldNot(HaveOccurred())
	pvcs := new(corev1.PersistentVolumeClaimList)
	err = ctrl.destClusterCli.List(ctx, pvcs)
	g.Expect(err).ShouldNot(HaveOccurred())
	g.Expect(len(pvcs.Items)).To(Equal(2))

	for _, pvc := range pvcs.Items {
		if pvc.Name == "test1" {
			t.Fatalf("ticdc pvc should deleted")
		}
	}
}

func TestCDCClusterID(t *testing.T) {
	ctl := new(ticdcCluster)
	ctl.provider = new(v1.Provider)
	ctl.provider.Spec.InstanceTypes = []v1.InstanceType{
		{Name: "c5-2xlarge"},
	}
	ctl.destCluster = &v1.Cluster{}
	ctl.destCluster.Name = "upstream"
	ctl.destNetwork = &v1.Network{
		ObjectMeta: metav1.ObjectMeta{Name: "n1"},
	}
	ctl.cdc = provider.NewGardenerCDCProvider()
	tc := new(v1alpha1.TidbCluster)
	cdc := new(v1.TiCDC)
	cdc.Spec.InstanceType = "c5-2xlarge"
	cdc.Spec.CDCClusterID = pointer.String("abcd")
	cdc.Spec.Replicas = 2
	cdc.Name = "1234"
	cdc.Namespace = "tidb"
	var gitops gitopsv2.DescribeClusterConfigurationFunc = func(cluster *v1.Cluster) (*v1.ClusterConfig, error) {
		return &v1.ClusterConfig{
			Spec: v1.ClusterConfigSpec{
				TiCDC: &v1.Configuration{
					Config: &v1.TomlConfig{
						MP: map[string]interface{}{},
					},
					BaseImage: pointer.String("gcr.io/pingcap-public/dbaas/ticdc"),
					Tag:       pointer.String("v6.2.0"),
				},
			},
		}, nil
	}
	require.Nil(t, ctl.overlayTicdc(tc, cdc, map[string]string{}, gitops))
	cfgValue, err := tc.Spec.TiCDC.Config.Get("cluster-id").AsString()
	require.Nil(t, err)
	require.Equal(t, "abcd", cfgValue)

	cdc.Spec.CDCClusterID = nil
	tc = new(v1alpha1.TidbCluster)
	require.Nil(t, ctl.overlayTicdc(tc, cdc, map[string]string{}, gitops))
	require.Nil(t, tc.Spec.TiCDC.Config.Get("cluster-id"))

	// cluster-id is not set when cdc version is 6.1
	cdc.Spec.CDCClusterID = pointer.String("abcdef")
	gitops = func(cluster *v1.Cluster) (*v1.ClusterConfig, error) {
		return &v1.ClusterConfig{
			Spec: v1.ClusterConfigSpec{
				TiCDC: &v1.Configuration{
					Config: &v1.TomlConfig{
						MP: map[string]interface{}{},
					},
					BaseImage: pointer.String("gcr.io/pingcap-public/dbaas/ticdc"),
					Tag:       pointer.String("v6.1.0"),
				},
			},
		}, nil
	}
	tc = new(v1alpha1.TidbCluster)
	require.Nil(t, ctl.overlayTicdc(tc, cdc, map[string]string{}, gitops))
	require.Nil(t, tc.Spec.TiCDC.Config.Get("cluster-id"))
}

func TestGetReferenceOwner(t *testing.T) {
	upstreamCluster := &v1.Cluster{}
	destCluster := &v1.Cluster{}
	require.Equal(t, destCluster, getReferenceOwner(upstreamCluster, destCluster))
	require.Equal(t, upstreamCluster, getReferenceOwner(upstreamCluster, nil))
}

func TestGetDestNamespace(t *testing.T) {
	upstreamCluster := &v1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "upstream",
		},
	}
	destCluster := &v1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "abcd",
		},
	}
	require.Equal(t, v1.ClusterPrefixId("abcd"), getDestNamespace(upstreamCluster, destCluster))
	require.Equal(t, v1.ClusterPrefixId("upstream")+"ticdc", getDestNamespace(upstreamCluster, nil))
}

func TestTicdcNdeGroupNamePrefix(t *testing.T) {
	require.Equal(t, nodegroup.NodeGroupNamePrefix("tidbupstream-cdc", string(v1.ComponentTiCDC)),
		ticdcNdeGroupNamePrefix(&v1.TiCDC{
			ObjectMeta: metav1.ObjectMeta{},
			Spec: v1.TiCDCSpec{
				UpstreamClusterName: "upstream",
			},
		}, nil))
	require.Equal(t, nodegroup.NodeGroupNamePrefix("tidbdownstream-cdc", string(v1.ComponentTiCDC)),
		ticdcNdeGroupNamePrefix(&v1.TiCDC{
			Spec: v1.TiCDCSpec{
				UpstreamClusterName: "upstream",
			},
		}, &v1.Cluster{
			ObjectMeta: metav1.ObjectMeta{
				Name: "downstream",
			},
		}))
	require.Equal(t, nodegroup.NodeGroupNamePrefix("tidbupstream", string(v1.ComponentTiCDC)),
		ticdcNdeGroupNamePrefix(&v1.TiCDC{
			Spec: v1.TiCDCSpec{
				ClusterName: "upstream",
			},
		}, &v1.Cluster{
			ObjectMeta: metav1.ObjectMeta{
				Name: "upstream",
			},
		}))
}

func TestGetCrossRegionTiDBClusterName(t *testing.T) {
	destCluster := &v1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "downstream",
		},
	}
	require.Equal(t, "downstream", getCrossRegionTiDBClusterName(destCluster))
	require.Equal(t, "", getCrossRegionTiDBClusterName(nil))
}