package cluster

import (
	"context"
	"testing"

	. "github.com/onsi/gomega"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes/scheme"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/controller-runtime/pkg/log"

	operatorv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/pingcap/label"
)

func TestReclaimPVs(t *testing.T) {
	cases := map[string]struct {
		cluster *cloudv1.Cluster
		objects []client.Object
		expect  func(*GomegaWithT, client.Client)
	}{
		"Basic": {
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "1001",
					Namespace: "test",
				},
			},
			objects: []client.Object{
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-tikv-0",
						Namespace: "tidb1001",
						Labels: map[string]string{
							label.InstanceLabelKey: "db",
						},
					},
				},
				&corev1.PersistentVolume{
					ObjectMeta: metav1.ObjectMeta{
						Name: "db-tikv-0",
						Labels: map[string]string{
							label.NamespaceLabelKey: "tidb1001",
						},
					},
					Spec: corev1.PersistentVolumeSpec{
						ClaimRef: &corev1.ObjectReference{
							Namespace: "tidb1001",
							Name:      "db-tikv-0",
						},
						PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain,
					},
				},
			},
			expect: func(g *GomegaWithT, cli client.Client) {
				pvc := &corev1.PersistentVolumeClaim{}
				err := cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0", Namespace: "tidb1001"}, pvc)
				g.Expect(err).NotTo(BeNil())
				g.Expect(apierrors.IsNotFound(err)).To(BeTrue())
				pv := &corev1.PersistentVolume{}
				err = cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0"}, pv)
				g.Expect(err).To(Succeed())
				g.Expect(pv.Spec.PersistentVolumeReclaimPolicy).To(Equal(corev1.PersistentVolumeReclaimDelete))
			},
		},
		"PV without label": {
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "1001",
					Namespace: "test",
				},
			},
			objects: []client.Object{
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-tikv-0",
						Namespace: "tidb1001",
						Labels: map[string]string{
							label.InstanceLabelKey: "db",
						},
					},
				},
				&corev1.PersistentVolume{
					ObjectMeta: metav1.ObjectMeta{
						Name: "db-tikv-0",
					},
					Spec: corev1.PersistentVolumeSpec{
						ClaimRef: &corev1.ObjectReference{
							Namespace: "tidb1001",
							Name:      "db-tikv-0",
						},
						PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain,
					},
				},
				&corev1.PersistentVolume{
					ObjectMeta: metav1.ObjectMeta{
						Name: "db-tikv-0-should-not-deleted",
					},
					Spec: corev1.PersistentVolumeSpec{
						ClaimRef: &corev1.ObjectReference{
							Namespace: "tidb1002",
							Name:      "db-tikv-0-should-not-deleted",
						},
						PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain,
					},
				},
			},
			expect: func(g *GomegaWithT, cli client.Client) {
				pvc := &corev1.PersistentVolumeClaim{}
				err := cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0", Namespace: "tidb1001"}, pvc)
				g.Expect(err).NotTo(BeNil())
				g.Expect(apierrors.IsNotFound(err)).To(BeTrue())
				pv := &corev1.PersistentVolume{}
				err = cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0"}, pv)
				g.Expect(err).To(Succeed())
				g.Expect(pv.Spec.PersistentVolumeReclaimPolicy).To(Equal(corev1.PersistentVolumeReclaimDelete))
				err = cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0-should-not-deleted"}, pv)
				g.Expect(err).To(Succeed())
				g.Expect(pv.Spec.PersistentVolumeReclaimPolicy).NotTo(Equal(corev1.PersistentVolumeReclaimDelete))
			},
		},
		"Skip PV without ClaimRef": {
			cluster: &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "1001",
					Namespace: "test",
				},
			},
			objects: []client.Object{
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-tikv-0",
						Namespace: "tidb1001",
						Labels: map[string]string{
							label.InstanceLabelKey: "db",
						},
					},
				},
				&corev1.PersistentVolume{
					ObjectMeta: metav1.ObjectMeta{
						Name: "db-tikv-0",
					},
					Spec: corev1.PersistentVolumeSpec{
						ClaimRef: &corev1.ObjectReference{
							Namespace: "tidb1001",
							Name:      "db-tikv-0",
						},
						PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain,
					},
				},
				&corev1.PersistentVolume{
					ObjectMeta: metav1.ObjectMeta{
						Name: "db-tikv-0-should-not-deleted",
					},
					Spec: corev1.PersistentVolumeSpec{
						ClaimRef: &corev1.ObjectReference{
							Namespace: "tidb1002",
							Name:      "db-tikv-0-should-not-deleted",
						},
						PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain,
					},
				},
				&corev1.PersistentVolume{
					ObjectMeta: metav1.ObjectMeta{
						Name: "db-tikv-0-without-claimref",
					},
					Spec: corev1.PersistentVolumeSpec{
						PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain,
					},
				},
			},
			expect: func(g *GomegaWithT, cli client.Client) {
				pvc := &corev1.PersistentVolumeClaim{}
				err := cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0", Namespace: "tidb1001"}, pvc)
				g.Expect(err).NotTo(BeNil())
				g.Expect(apierrors.IsNotFound(err)).To(BeTrue())
				pv := &corev1.PersistentVolume{}
				err = cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0"}, pv)
				g.Expect(err).To(Succeed())
				g.Expect(pv.Spec.PersistentVolumeReclaimPolicy).To(Equal(corev1.PersistentVolumeReclaimDelete))
				err = cli.Get(context.TODO(), types.NamespacedName{Name: "db-tikv-0-should-not-deleted"}, pv)
				g.Expect(err).To(Succeed())
				g.Expect(pv.Spec.PersistentVolumeReclaimPolicy).NotTo(Equal(corev1.PersistentVolumeReclaimDelete))
			},
		},
	}
	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			g := NewGomegaWithT(t)
			s := runtime.NewScheme()
			g.Expect(scheme.AddToScheme(s)).To(Succeed())
			g.Expect(operatorv1alpha1.AddToScheme(s)).To(Succeed())
			cli := fake.NewClientBuilder().WithScheme(s).WithObjects(tc.objects...).Build()
			ctx := context.Background()
			rx := &reconcileContext{
				ctx:         ctx,
				shootClient: cli,
				ClusterReconciler: &ClusterReconciler{
					Log: log.NullLogger{},
				},
			}
			rx.Target = tc.cluster
			rx.GitOps, rx.Cluster = rx, rx
			g.Expect(rx.reclaimPVs()).To(Succeed())
			tc.expect(g, cli)
		})
	}
}