package metricsprovider

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/cluster-service-ng/pkg/k8s"
	autoscalingv1 "k8s.io/api/autoscaling/v1"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes/scheme"
	scalefake "k8s.io/client-go/scale/fake"
	core "k8s.io/client-go/testing"
	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
	metricsfake "k8s.io/metrics/pkg/client/clientset/versioned/fake"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
)

type extAvgMemoryTestCase struct {
	memoryRequests           []string
	memoryLimits             []string
	reportedMemory           []string
	podPhase                 []corev1.PodPhase
	podReadiness             []corev1.ConditionStatus
	podGracefulShutdownBegin []bool
	expectedValue            int64
	expectedError            string
}

func (tc *extAvgMemoryTestCase) buildHPA() *autoscalingv2.HorizontalPodAutoscaler {
	return &autoscalingv2.HorizontalPodAutoscaler{
		ObjectMeta: metav1.ObjectMeta{
			Name:      testHPAName,
			Namespace: testNamespace,
		},
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
				APIVersion: "apps/v1",
				Kind:       "Deployment",
				Name:       testDeploymentName,
			},
			MinReplicas: ptr.To(int32(1)),
			MaxReplicas: 10,
			Metrics: []autoscalingv2.MetricSpec{
				{
					Type: autoscalingv2.ObjectMetricSourceType,
					Object: &autoscalingv2.ObjectMetricSource{
						DescribedObject: autoscalingv2.CrossVersionObjectReference{
							APIVersion: "autoscaling/v2",
							Kind:       "HorizontalPodAutoscaler",
							Name:       testHPAName,
						},
						Metric: autoscalingv2.MetricIdentifier{
							Name: MetricNameExtAvgMemoryUtilization,
						},
						Target: autoscalingv2.MetricTarget{
							Type:  autoscalingv2.ValueMetricType,
							Value: resource.NewQuantity(70, resource.DecimalSI),
						},
					},
				},
			},
		},
	}
}

func (tc *extAvgMemoryTestCase) buildPods() []client.Object {
	numPods := len(tc.memoryLimits)
	if len(tc.memoryRequests) > numPods {
		numPods = len(tc.memoryRequests)
	}
	if len(tc.reportedMemory) > numPods {
		numPods = len(tc.reportedMemory)
	}

	pods := make([]client.Object, 0, numPods)
	for i := 0; i < numPods; i++ {
		resources := corev1.ResourceRequirements{
			Requests: corev1.ResourceList{},
			Limits:   corev1.ResourceList{},
		}
		if i < len(tc.memoryRequests) && tc.memoryRequests[i] != "" {
			resources.Requests[corev1.ResourceMemory] = resource.MustParse(tc.memoryRequests[i])
		}
		if i < len(tc.memoryLimits) && tc.memoryLimits[i] != "" {
			resources.Limits[corev1.ResourceMemory] = resource.MustParse(tc.memoryLimits[i])
		}

		phase := corev1.PodRunning
		if tc.podPhase != nil && i < len(tc.podPhase) {
			phase = tc.podPhase[i]
		}

		podReadiness := corev1.ConditionTrue
		if tc.podReadiness != nil && i < len(tc.podReadiness) {
			podReadiness = tc.podReadiness[i]
		}

		podMeta := metav1.ObjectMeta{
			Name:      fmt.Sprintf("%s-%d", testPodNamePrefix, i),
			Namespace: testNamespace,
			Labels: map[string]string{
				"app": testDeploymentName,
			},
		}
		if tc.podGracefulShutdownBegin != nil && i < len(tc.podGracefulShutdownBegin) && tc.podGracefulShutdownBegin[i] {
			podMeta.Annotations = map[string]string{
				v1alpha1.AnnoKeyTiProxyGracefulShutdownBeginTime: metav1.Now().Format(time.RFC3339),
			}
		}

		pod := &corev1.Pod{
			ObjectMeta: podMeta,
			Spec: corev1.PodSpec{
				Containers: []corev1.Container{
					{
						Name:      "main",
						Resources: resources,
					},
				},
			},
			Status: corev1.PodStatus{
				Phase: phase,
				Conditions: []corev1.PodCondition{
					{
						Type:   corev1.PodReady,
						Status: podReadiness,
					},
				},
			},
		}
		pods = append(pods, pod)
	}
	return pods
}

func (tc *extAvgMemoryTestCase) buildPodMetrics() *metricsapi.PodMetricsList {
	metrics := &metricsapi.PodMetricsList{}
	for i, mem := range tc.reportedMemory {
		podMetric := metricsapi.PodMetrics{
			ObjectMeta: metav1.ObjectMeta{
				Name:      fmt.Sprintf("%s-%d", testPodNamePrefix, i),
				Namespace: testNamespace,
				Labels: map[string]string{
					"app": testDeploymentName,
				},
			},
			Timestamp: metav1.Now(),
			Window:    metav1.Duration{},
			Containers: []metricsapi.ContainerMetrics{
				{
					Name: "main",
					Usage: corev1.ResourceList{
						corev1.ResourceMemory: resource.MustParse(mem),
					},
				},
			},
		}
		metrics.Items = append(metrics.Items, podMetric)
	}
	return metrics
}

func (tc *extAvgMemoryTestCase) buildScale() *autoscalingv1.Scale {
	return &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{
			Name:      testDeploymentName,
			Namespace: testNamespace,
		},
		Spec: autoscalingv1.ScaleSpec{
			Replicas: int32(len(tc.memoryLimits)),
		},
		Status: autoscalingv1.ScaleStatus{
			Replicas: int32(len(tc.memoryLimits)),
			Selector: "app=" + testDeploymentName,
		},
	}
}

func (tc *extAvgMemoryTestCase) setupProvider(t *testing.T) provider.MetricsProvider {
	t.Helper()

	s := runtime.NewScheme()
	require.NoError(t, scheme.AddToScheme(s))
	require.NoError(t, autoscalingv2.AddToScheme(s))

	objects := tc.buildPods()
	objects = append(objects, tc.buildHPA())
	kubeClient := fake.NewClientBuilder().
		WithScheme(s).
		WithObjects(objects...).
		Build()

	fakeMetricsClientset := &metricsfake.Clientset{}
	fakeMetricsClientset.AddReactor("list", "pods", func(action core.Action) (bool, runtime.Object, error) {
		return true, tc.buildPodMetrics(), nil
	})

	fakeScaleClient := &scalefake.FakeScaleClient{}
	fakeScaleClient.AddReactor("get", "deployments", func(action core.Action) (bool, runtime.Object, error) {
		return true, tc.buildScale(), nil
	})

	return NewCMProvider(kubeClient, k8s.RESTMapper(), 0, fakeMetricsClientset.MetricsV1beta1(), fakeScaleClient)
}

func (tc *extAvgMemoryTestCase) runTest(t *testing.T) {
	t.Helper()

	p := tc.setupProvider(t)
	info := provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{
			Group:    "autoscaling",
			Resource: "horizontalpodautoscalers",
		},
		Metric:     MetricNameExtAvgMemoryUtilization,
		Namespaced: true,
	}

	result, err := p.GetMetricByName(
		context.Background(),
		types.NamespacedName{Namespace: testNamespace, Name: testHPAName},
		info,
		labels.Everything(),
	)

	if tc.expectedError != "" {
		require.Error(t, err)
		assert.Contains(t, err.Error(), tc.expectedError)
		return
	}

	require.NoError(t, err)
	require.NotNil(t, result)
	assert.Equal(t, tc.expectedValue, result.Value.Value())
}

func TestExtAvgMemoryUtilization_UsesLimitsAsDenominator(t *testing.T) {
	tc := extAvgMemoryTestCase{
		memoryRequests: []string{"1Gi", "1Gi"},
		memoryLimits:   []string{"2Gi", "2Gi"},
		reportedMemory: []string{"1Gi", "1Gi"},
		expectedValue:  50,
	}
	tc.runTest(t)
}

func TestExtAvgMemoryUtilization_IgnoresFailedPods(t *testing.T) {
	tc := extAvgMemoryTestCase{
		memoryRequests: []string{"1Gi", "1Gi"},
		memoryLimits:   []string{"2Gi", "2Gi"},
		reportedMemory: []string{"1Gi", "2Gi"},
		podPhase:       []corev1.PodPhase{corev1.PodRunning, corev1.PodFailed},
		expectedValue:  50,
	}
	tc.runTest(t)
}

func TestExtAvgMemoryUtilization_IgnoresPendingPodLimits(t *testing.T) {
	tc := extAvgMemoryTestCase{
		memoryRequests: []string{"1Gi", "1Gi"},
		memoryLimits:   []string{"1Gi", "10Gi"},
		reportedMemory: []string{"512Mi", "10Gi"},
		podPhase:       []corev1.PodPhase{corev1.PodRunning, corev1.PodPending},
		expectedValue:  50,
	}
	tc.runTest(t)
}

func TestExtAvgMemoryUtilization_IgnoresDrainingPods(t *testing.T) {
	tc := extAvgMemoryTestCase{
		memoryRequests:           []string{"1Gi", "1Gi"},
		memoryLimits:             []string{"2Gi", "2Gi"},
		reportedMemory:           []string{"1Gi", "2Gi"},
		podGracefulShutdownBegin: []bool{false, true},
		expectedValue:            50,
	}
	tc.runTest(t)
}

func TestExtAvgMemoryUtilization_MissingLimit(t *testing.T) {
	tc := extAvgMemoryTestCase{
		memoryRequests: []string{"1Gi"},
		memoryLimits:   []string{""},
		reportedMemory: []string{"512Mi"},
		expectedError:  "missing limit for memory",
	}
	tc.runTest(t)
}

func TestExtAvgMemoryUtilization_NoMetrics(t *testing.T) {
	tc := extAvgMemoryTestCase{
		memoryRequests: []string{"1Gi"},
		memoryLimits:   []string{"2Gi"},
		reportedMemory: []string{},
		expectedError:  "no metrics",
	}
	tc.runTest(t)
}

func TestListAllMetricsIncludesExtAvgMemoryUtilization(t *testing.T) {
	p := &cmProvider{}
	for _, metric := range p.ListAllMetrics() {
		if metric.Metric == MetricNameExtAvgMemoryUtilization {
			assert.Equal(t, schema.GroupResource{
				Group:    "autoscaling",
				Resource: "horizontalpodautoscalers",
			}, metric.GroupResource)
			assert.True(t, metric.Namespaced)
			return
		}
	}
	t.Fatalf("metric %s is not listed", MetricNameExtAvgMemoryUtilization)
}