package metricsprovider

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/prometheus/common/expfmt"
	"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"
	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"
	metricsfake "k8s.io/metrics/pkg/client/clientset/versioned/fake"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
)

const (
	bwTestNamespace        = "tidb12345"
	bwTestHPAName          = "tiproxy-hpa"
	bwTestGroupName        = "db-comp-abc-tiproxy"
	bwTestNodeName         = "ip-10-0-1-100.ec2.internal"
	bwTestNodeIP           = "10.0.1.100"
	bwTestNodeName2        = "ip-10-0-1-200.ec2.internal"
	bwTestNodeIP2          = "10.0.1.200"
	bwTestNodeExporterPort = int32(9100)
)

func buildBWTestScheme(t *testing.T) *runtime.Scheme {
	t.Helper()
	s := runtime.NewScheme()
	require.NoError(t, scheme.AddToScheme(s))
	require.NoError(t, autoscalingv2.AddToScheme(s))
	require.NoError(t, v1alpha1.Install(s))
	return s
}

func buildBWTestHPA() *autoscalingv2.HorizontalPodAutoscaler {
	return &autoscalingv2.HorizontalPodAutoscaler{
		ObjectMeta: metav1.ObjectMeta{
			Name:      bwTestHPAName,
			Namespace: bwTestNamespace,
		},
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			ScaleTargetRef: autoscalingv2.CrossVersionObjectReference{
				APIVersion: "core.pingcap.com/v1alpha1",
				Kind:       "TiProxyGroup",
				Name:       bwTestGroupName,
			},
			MinReplicas: ptr.To(int32(1)),
			MaxReplicas: 5,
		},
	}
}

func buildBWTestTiProxyGroup() *v1alpha1.TiProxyGroup {
	return &v1alpha1.TiProxyGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:      bwTestGroupName,
			Namespace: bwTestNamespace,
		},
	}
}

func buildBWScale() *autoscalingv1.Scale {
	return &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{
			Name:      bwTestGroupName,
			Namespace: bwTestNamespace,
		},
		Spec:   autoscalingv1.ScaleSpec{Replicas: 1},
		Status: autoscalingv1.ScaleStatus{Replicas: 1, Selector: "app.kubernetes.io/instance=" + bwTestGroupName},
	}
}

func buildBWTestPod(name, nodeName, nodeIP string) *corev1.Pod {
	return buildBWTestPodWithReady(name, nodeName, nodeIP, corev1.ConditionTrue)
}

func buildBWTestPodWithReady(name, nodeName, nodeIP string, ready corev1.ConditionStatus) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: bwTestNamespace,
			Labels:    map[string]string{"app.kubernetes.io/instance": bwTestGroupName},
		},
		Spec: corev1.PodSpec{NodeName: nodeName},
		Status: corev1.PodStatus{
			Phase:  corev1.PodRunning,
			HostIP: nodeIP,
			Conditions: []corev1.PodCondition{
				{
					Type:   corev1.PodReady,
					Status: ready,
				},
			},
		},
	}
}

func srvHostPort(srv *httptest.Server) (host string, port int32) {
	h, p, _ := net.SplitHostPort(strings.TrimPrefix(srv.URL, "http://"))
	n, _ := strconv.ParseInt(p, 10, 32)
	return h, int32(n)
}

func buildNodeExporterPod(nodeName, hostIP string, port int32) *corev1.Pod {
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      fmt.Sprintf("node-exporter-%s", nodeName),
			Namespace: nodeExporterNamespace,
			Labels:    map[string]string{nodeExporterInstanceKey: nodeExporterInstanceValue},
		},
		Spec: corev1.PodSpec{
			NodeName: nodeName,
			Containers: []corev1.Container{
				{
					Name:  "node-exporter",
					Ports: []corev1.ContainerPort{{Name: nodeExporterPortName, ContainerPort: port}},
				},
			},
		},
		Status: corev1.PodStatus{
			Phase:  corev1.PodRunning,
			HostIP: hostIP,
		},
	}
}

func buildAWSNode(nodeName, instanceType string) *corev1.Node {
	return &corev1.Node{
		ObjectMeta: metav1.ObjectMeta{
			Name: nodeName,
			Labels: map[string]string{
				"eks.amazonaws.com/nodegroup":      "default",
				"node.kubernetes.io/instance-type": instanceType,
			},
		},
	}
}

func buildAlicloudNode(nodeName, instanceType string) *corev1.Node {
	return &corev1.Node{
		ObjectMeta: metav1.ObjectMeta{
			Name: nodeName,
			Labels: map[string]string{
				"ack.aliyun.com":                   "some-cluster-id",
				"node.kubernetes.io/instance-type": instanceType,
			},
		},
	}
}

func netdevMetricsBody(rxBytes, txBytes float64, device string) string {
	const speedBytes = 1.25e9
	return fmt.Sprintf(`# HELP node_network_receive_bytes_total Network device statistic receive_bytes.
# TYPE node_network_receive_bytes_total counter
node_network_receive_bytes_total{device="%s"} %g
# HELP node_network_transmit_bytes_total Network device statistic transmit_bytes.
# TYPE node_network_transmit_bytes_total counter
node_network_transmit_bytes_total{device="%s"} %g
# HELP node_network_speed_bytes Network device property: speed_bytes
# TYPE node_network_speed_bytes gauge
node_network_speed_bytes{device="%s"} %g
`, device, rxBytes, device, txBytes, device, speedBytes)
}

type bwProviderFixture struct {
	kubeObjects     []runtime.Object
	nodeExporterSrv *httptest.Server
	scaleReactor    func(action core.Action) (bool, runtime.Object, error)
	prewarmCache    map[string]networkCounterSample
}

func (f *bwProviderFixture) build(t *testing.T) *cmProvider {
	t.Helper()
	s := buildBWTestScheme(t)

	builder := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(buildBWTestTiProxyGroup())
	for _, o := range f.kubeObjects {
		builder.WithRuntimeObjects(o)
	}
	kubeClient := builder.Build()

	fakeMetricsClientset := &metricsfake.Clientset{}
	fakeScaleClient := &scalefake.FakeScaleClient{}
	if f.scaleReactor != nil {
		fakeScaleClient.AddReactor("get", "*", f.scaleReactor)
	} else {
		fakeScaleClient.AddReactor("get", "*", func(action core.Action) (bool, runtime.Object, error) {
			return true, buildBWScale(), nil
		})
	}

	mapper := k8s.RESTMapper()
	p := &cmProvider{
		kubeClient:             kubeClient,
		mapper:                 mapper,
		disableScalueIn:        0,
		resourceMetricsClient:  fakeMetricsClientset.MetricsV1beta1(),
		scaleClient:            fakeScaleClient,
		bwCache:                newBandwidthCache(),
		nodeExporterHTTPClient: f.nodeExporterSrv.Client(),
	}

	if f.prewarmCache != nil {
		p.bwCache.mu.Lock()
		for k, v := range f.prewarmCache {
			p.bwCache.items[k] = v
		}
		p.bwCache.mu.Unlock()
	}

	return p
}

func bwMetricInfo() provider.CustomMetricInfo {
	return provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{
			Group:    "core.pingcap.com",
			Resource: "tiproxygroups",
		},
		Metric:     MetricNameExtAvgBandwidthUtilization,
		Namespaced: true,
	}
}

func bwTiProxyGroupName() types.NamespacedName {
	return types.NamespacedName{Namespace: bwTestNamespace, Name: bwTestGroupName}
}

func bwTargetName() types.NamespacedName {
	return bwTiProxyGroupName()
}

func TestExtAvgBandwidthUtilization_WarmUp(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprint(w, netdevMetricsBody(1e9, 5e8, "eth0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "warm-up")
}

func TestExtAvgBandwidthUtilization_SecondScrapeComputes(t *testing.T) {

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprint(w, netdevMetricsBody(2e9, 1e9, "eth0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-15 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {
				timestamp: prevTime,
				rxBytes:   1e9,
				txBytes:   5e8,
			},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)

	// deltaRx = 2e9 - 1e9 = 1e9 bytes over ~15 s → rx_rate ≈ 66.7 MB/s
	// deltaTx = 1e9 - 5e8 = 5e8 bytes over ~15 s → tx_rate ≈ 33.3 MB/s
	// dominant = rx_rate; util = rx_rate / 234_375_000 (m8g.xlarge baseline) ≈ 28%
	// Allow ±10 pct due to non-exact elapsed time in tests.
	got := result.Value.Value()
	assert.InDelta(t, 28, got, 10, "expected ~28%% utilization, got %d", got)
}

func TestExtAvgBandwidthUtilization_TwoPodsAveraged(t *testing.T) {

	calls := 0
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls++
		if calls == 1 {
			_, _ = fmt.Fprint(w, netdevMetricsBody(2e9, 0, "eth0"))
		} else {
			_, _ = fmt.Fprint(w, netdevMetricsBody(3e9, 0, "eth0"))
		}
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	scale := &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{Name: bwTestGroupName, Namespace: bwTestNamespace},
		Spec:       autoscalingv1.ScaleSpec{Replicas: 2},
		Status:     autoscalingv1.ScaleStatus{Replicas: 2, Selector: "app.kubernetes.io/instance=" + bwTestGroupName},
	}

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildBWTestPod("pod-1", bwTestNodeName2, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildNodeExporterPod(bwTestNodeName2, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
			buildAWSNode(bwTestNodeName2, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		scaleReactor: func(action core.Action) (bool, runtime.Object, error) {
			return true, scale, nil
		},
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0":  {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
			targetKey + "/" + bwTestNodeName2 + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	got := result.Value.Value()
	// node1: Δrx=1e9/10s=1e8 B/s → 42.7% of 234_375_000 (m8g.xlarge baseline)
	// node2: Δrx=2e9/10s=2e8 B/s → 85.3% of 234_375_000
	// avg ≈ 64%
	assert.InDelta(t, 64, got, 15, "expected ~64%% average, got %d", got)
}

func TestExtAvgBandwidthUtilization_IgnoresDrainingPods(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprint(w, netdevMetricsBody(2e9, 0, "eth0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	scale := &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{Name: bwTestGroupName, Namespace: bwTestNamespace},
		Spec:       autoscalingv1.ScaleSpec{Replicas: 2},
		Status:     autoscalingv1.ScaleStatus{Replicas: 2, Selector: "app.kubernetes.io/instance=" + bwTestGroupName},
	}

	drainingPod := buildBWTestPod("pod-1", bwTestNodeName2, srvHost)
	drainingPod.Annotations = map[string]string{
		v1alpha1.AnnoKeyTiProxyGracefulShutdownBeginTime: time.Now().Format(time.RFC3339),
	}

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			drainingPod,
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		scaleReactor: func(action core.Action) (bool, runtime.Object, error) {
			return true, scale, nil
		},
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	got := result.Value.Value()
	assert.InDelta(t, 43, got, 15, "expected only non-draining pod to count, got %d", got)
}

func TestExtAvgBandwidthUtilization_TwoPodsOnSameNodeCountTwice(t *testing.T) {
	calls := 0
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls++
		switch calls {
		case 1:
			_, _ = fmt.Fprint(w, netdevMetricsBody(3e9, 0, "eth0"))
		case 2:
			_, _ = fmt.Fprint(w, netdevMetricsBody(1.5e9, 0, "eth0"))
		default:
			t.Fatalf("unexpected scrape #%d", calls)
		}
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	scale := &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{Name: bwTestGroupName, Namespace: bwTestNamespace},
		Spec:       autoscalingv1.ScaleSpec{Replicas: 3},
		Status:     autoscalingv1.ScaleStatus{Replicas: 3, Selector: "app.kubernetes.io/instance=" + bwTestGroupName},
	}

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildBWTestPod("pod-1", bwTestNodeName, srvHost),
			buildBWTestPod("pod-2", bwTestNodeName2, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildNodeExporterPod(bwTestNodeName2, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
			buildAWSNode(bwTestNodeName2, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		scaleReactor: func(action core.Action) (bool, runtime.Object, error) {
			return true, scale, nil
		},
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0":  {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
			targetKey + "/" + bwTestNodeName2 + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	assert.Equal(t, 2, calls, "should scrape each unique node once")

	got := result.Value.Value()
	// node1: Δrx=2e9/10s=2e8 B/s → 85.3%, counted for two pods
	// node2: Δrx=5e8/10s=5e7 B/s → 21.3%, counted for one pod
	// weighted avg across pods ≈ 64%
	assert.InDelta(t, 64, got, 10, "expected same-node pods to be weighted twice, got %d", got)
}

func TestExtAvgBandwidthUtilization_TxDominatesFleetWide(t *testing.T) {
	calls := 0
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls++
		if calls == 1 {
			_, _ = fmt.Fprint(w, netdevMetricsBody(1e9, 3e9, "eth0"))
		} else {
			_, _ = fmt.Fprint(w, netdevMetricsBody(2e9, 4e9, "eth0"))
		}
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	scale := &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{Name: bwTestGroupName, Namespace: bwTestNamespace},
		Spec:       autoscalingv1.ScaleSpec{Replicas: 2},
		Status:     autoscalingv1.ScaleStatus{Replicas: 2, Selector: "app.kubernetes.io/instance=" + bwTestGroupName},
	}

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildBWTestPod("pod-1", bwTestNodeName2, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildNodeExporterPod(bwTestNodeName2, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
			buildAWSNode(bwTestNodeName2, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		scaleReactor: func(action core.Action) (bool, runtime.Object, error) {
			return true, scale, nil
		},
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0":  {timestamp: prevTime, rxBytes: 0, txBytes: 1e9},
			targetKey + "/" + bwTestNodeName2 + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 2e9},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	got := result.Value.Value()
	// node0: Δrx=1e9/10s=1e8 B/s → rxUtil=42.7%,  Δtx=2e9/10s=2e8 B/s → txUtil=85.3%
	// node1: Δrx=1e9/10s=1e8 B/s → rxUtil=42.7%,  Δtx=2e9/10s=2e8 B/s → txUtil=85.3%
	// avgRx = 42.7%,  avgTx = 85.3%
	// result = max(42.7, 85.3) = 85.3%  →  tx dominates
	assert.InDelta(t, 85, got, 10, "tx should dominate fleet-wide, expected ~85%%, got %d", got)
}

func TestExtAvgBandwidthUtilization_CounterReset(t *testing.T) {

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprint(w, netdevMetricsBody(100, 0, "eth0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-15 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {
				timestamp: prevTime,
				rxBytes:   9e11,
				txBytes:   0,
			},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	assert.Equal(t, int64(0), result.Value.Value(), "counter reset should yield 0 utilization")
}

func TestExtAvgBandwidthUtilization_NoETH0Interface(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprint(w, netdevMetricsBody(1e9, 0, "eni0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-15 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eni0": {timestamp: prevTime, rxBytes: 0, txBytes: 0},
		},
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "no usable bandwidth samples")
}

func TestExtAvgBandwidthUtilization_UnknownProvider(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_, _ = w.Write(nil)
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	unknownNode := &corev1.Node{
		ObjectMeta: metav1.ObjectMeta{
			Name:   bwTestNodeName,
			Labels: map[string]string{"kubernetes.io/os": "linux"},
		},
	}

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			unknownNode,
		},
		nodeExporterSrv: srv,
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "unsupported cloud provider")
}

func TestExtAvgBandwidthUtilization_WrongResourceType(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {}))
	defer srv.Close()

	f := &bwProviderFixture{
		kubeObjects:     []runtime.Object{buildBWTestHPA()},
		nodeExporterSrv: srv,
	}
	p := f.build(t)

	wrongInfo := provider.CustomMetricInfo{
		GroupResource: schema.GroupResource{
			Group:    "autoscaling",
			Resource: "horizontalpodautoscalers",
		},
		Metric:     MetricNameExtAvgBandwidthUtilization,
		Namespaced: true,
	}

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTiProxyGroupName(), wrongInfo, labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "not supported")
}

func TestExtAvgBandwidthUtilization_UsesTiProxyGroupScaleRef(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_, _ = fmt.Fprint(w, netdevMetricsBody(2e9, 1e9, "eth0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTiProxyGroupName().String()
	prevTime := time.Now().Add(-15 * time.Second)
	var scaleName string
	var scaleResource schema.GroupVersionResource

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		scaleReactor: func(action core.Action) (bool, runtime.Object, error) {
			getAction, ok := action.(core.GetAction)
			require.True(t, ok)
			scaleName = getAction.GetName()
			scaleResource = action.GetResource()
			return true, buildBWScale(), nil
		},
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {
				timestamp: prevTime,
				rxBytes:   1e9,
				txBytes:   5e8,
			},
		},
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTiProxyGroupName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	assert.Equal(t, bwTestGroupName, scaleName)
	assert.Equal(t, "core.pingcap.com", scaleResource.Group)
	assert.Equal(t, "tiproxygroups", scaleResource.Resource)
}

func TestExtAvgBandwidthUtilization_ScrapeURLIncludesNetclass(t *testing.T) {
	var capturedURL string
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		capturedURL = r.URL.RawQuery
		_, _ = fmt.Fprint(w, netdevMetricsBody(1e9, 5e8, "eth0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
	}
	p := f.build(t)

	_, _ = p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	assert.Contains(t, capturedURL, "collect[]=netdev")
	assert.Contains(t, capturedURL, "collect[]=netclass")
}

func TestExtAvgBandwidthUtilization_AWSSecondScrapeComputes(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprintf(w, `# TYPE node_network_receive_bytes_total counter
node_network_receive_bytes_total{device="eth0"} 2e9
# TYPE node_network_transmit_bytes_total counter
node_network_transmit_bytes_total{device="eth0"} 1e9
# TYPE node_network_speed_bytes gauge
node_network_speed_bytes{device="eni557e1959980"} 1.25e+09
`)
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-15 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {
				timestamp: prevTime,
				rxBytes:   1e9,
				txBytes:   5e8,
			},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	got := result.Value.Value()
	// deltaRx = 1e9 bytes over ~15 s → rx_rate ≈ 66.7 MB/s
	// m8g.xlarge baseline = 234_375_000 B/s (1.875 Gbit/s)
	// util ≈ 66.7/234.4 ≈ 28.4%
	assert.InDelta(t, 28, got, 10, "expected ~28%% utilization, got %d", got)
}

func TestExtAvgBandwidthUtilization_AWSUnknownInstanceType(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName2, srvHost),
			buildNodeExporterPod(bwTestNodeName2, srvHost, srvPort),
			buildAWSNode(bwTestNodeName2, "m99z.unknown"),
		},
		nodeExporterSrv: srv,
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "unknown aws instance type")
}

func TestExtAvgBandwidthUtilization_AWSUsesBaselineNotBurstSpeed(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprintf(w, `# TYPE node_network_receive_bytes_total counter
node_network_receive_bytes_total{device="eth0"} 2e9
# TYPE node_network_transmit_bytes_total counter
node_network_transmit_bytes_total{device="eth0"} 0
# TYPE node_network_speed_bytes gauge
node_network_speed_bytes{device="eni557e1959980"} 1.5625e9
`)
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	got := result.Value.Value()
	// rx_rate = 1e9/10s = 1e8 B/s
	// node_network_speed_bytes reports 1.5625e9 (burst), but we use baseline: 234_375_000 (1.875 Gbit/s)
	// util against baseline = 1e8/234_375_000 ≈ 42.7%
	// util against burst    = 1e8/1.5625e9    ≈  6.4%  (wrong, should not be used)
	assert.InDelta(t, 42, got, 10, "should use baseline speed (234MB/s), not burst speed (1562MB/s)")
}

func TestExtAvgBandwidthUtilization_AlicloudSecondScrapeComputes(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprintf(w, `# TYPE node_network_receive_bytes_total counter
node_network_receive_bytes_total{device="eth0"} 2e9
# TYPE node_network_transmit_bytes_total counter
node_network_transmit_bytes_total{device="eth0"} 1e9
# TYPE node_network_speed_bytes gauge
node_network_speed_bytes{device="eth0"} -125000
`)
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-15 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAlicloudNode(bwTestNodeName, "ecs.g9i.large"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {
				timestamp: prevTime,
				rxBytes:   1e9,
				txBytes:   5e8,
			},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	got := result.Value.Value()
	// deltaRx = 1e9 bytes over ~15 s → rx_rate ≈ 66.7 MB/s
	// ecs.g9i.large speed = 320_000_000 B/s
	// util ≈ 66.7/320 ≈ 20.8%
	assert.InDelta(t, 20, got, 10, "expected ~20%% utilization, got %d", got)
}

func TestExtAvgBandwidthUtilization_AlicloudUnknownInstanceType(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName2, srvHost),
			buildNodeExporterPod(bwTestNodeName2, srvHost, srvPort),
			buildAlicloudNode(bwTestNodeName2, "ecs.unknown.large"),
		},
		nodeExporterSrv: srv,
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "unknown alicloud instance type")
}

func TestExtAvgBandwidthUtilization_AlicloudIgnoresNegativeSpeed(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprintf(w, `# TYPE node_network_receive_bytes_total counter
node_network_receive_bytes_total{device="eth0"} 2e9
# TYPE node_network_transmit_bytes_total counter
node_network_transmit_bytes_total{device="eth0"} 0
# TYPE node_network_speed_bytes gauge
node_network_speed_bytes{device="eth0"} -125000
`)
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAlicloudNode(bwTestNodeName, "ecs.g9i.large"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	// Negative node_network_speed_bytes is ignored; hardcoded 320MB/s is used instead.
	// rx_rate = 1e9/10s = 1e8 B/s; util = 1e8/320e6 ≈ 31%
	assert.InDelta(t, 31, result.Value.Value(), 10)
}

func TestExtAvgBandwidthUtilization_AlicloudENIDeviceIgnored(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprintf(w, `# TYPE node_network_receive_bytes_total counter
node_network_receive_bytes_total{device="eth0"} 2e9
node_network_receive_bytes_total{device="eth1"} 5e8
# TYPE node_network_transmit_bytes_total counter
node_network_transmit_bytes_total{device="eth0"} 0
node_network_transmit_bytes_total{device="eth1"} 0
# TYPE node_network_speed_bytes gauge
node_network_speed_bytes{device="eth0"} -125000
node_network_speed_bytes{device="eth1"} -125000
`)
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAlicloudNode(bwTestNodeName, "ecs.g9i.large"),
		},
		nodeExporterSrv: srv,
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
			targetKey + "/" + bwTestNodeName + "/eth1": {timestamp: prevTime, rxBytes: 2e8, txBytes: 0},
		},
	}
	p := f.build(t)

	result, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.NoError(t, err)
	require.NotNil(t, result)
	// eth0: rx_rate = 1e9/10s = 1e8 B/s → util = 1e8/320e6 ≈ 31%
	// eth1: rx_rate = 3e8/10s = 3e7 B/s → util = 3e7/320e6 ≈ 9%
	// max across eth0+eth1 = 31%
	assert.InDelta(t, 31, result.Value.Value(), 10)
}

func TestExtAvgBandwidthUtilization_NoRunningPods(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {}))
	defer srv.Close()

	pendingPod := buildBWTestPod("pod-0", bwTestNodeName, bwTestNodeIP)
	pendingPod.Status.Phase = corev1.PodPending

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			pendingPod,
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "no running tiproxy pods")
}

func TestExtAvgBandwidthUtilization_QuorumGuard(t *testing.T) {
	const bwTestNodeName3 = "ip-10-0-1-300.ec2.internal"

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = fmt.Fprint(w, netdevMetricsBody(2e9, 0, "eth0"))
	}))
	defer srv.Close()
	srvHost, srvPort := srvHostPort(srv)

	targetKey := bwTargetName().String()
	prevTime := time.Now().Add(-10 * time.Second)

	scale := &autoscalingv1.Scale{
		ObjectMeta: metav1.ObjectMeta{Name: bwTestGroupName, Namespace: bwTestNamespace},
		Spec:       autoscalingv1.ScaleSpec{Replicas: 3},
		Status:     autoscalingv1.ScaleStatus{Replicas: 3, Selector: "app.kubernetes.io/instance=" + bwTestGroupName},
	}

	f := &bwProviderFixture{
		kubeObjects: []runtime.Object{
			buildBWTestHPA(),
			buildBWTestPod("pod-0", bwTestNodeName, srvHost),
			buildBWTestPod("pod-1", bwTestNodeName2, "192.168.99.99"),
			buildBWTestPod("pod-2", bwTestNodeName3, "192.168.99.100"),
			buildNodeExporterPod(bwTestNodeName, srvHost, srvPort),
			buildAWSNode(bwTestNodeName, "m8g.xlarge"),
			buildAWSNode(bwTestNodeName2, "m8g.xlarge"),
			buildAWSNode(bwTestNodeName3, "m8g.xlarge"),
		},
		nodeExporterSrv: srv,
		scaleReactor: func(action core.Action) (bool, runtime.Object, error) {
			return true, scale, nil
		},
		prewarmCache: map[string]networkCounterSample{
			targetKey + "/" + bwTestNodeName + "/eth0": {timestamp: prevTime, rxBytes: 1e9, txBytes: 0},
		},
	}
	p := f.build(t)

	_, err := p.GetMetricExtAvgBandwidthUtilization(
		context.Background(), bwTargetName(), bwMetricInfo(), labels.Everything())

	require.Error(t, err)
	assert.Contains(t, err.Error(), "too few usable bandwidth samples")
}

func TestDtoFamilyToDeviceValues_NilFamily(t *testing.T) {
	result := dtoFamilyToDeviceValues(nil)
	assert.Empty(t, result)
}

func TestDtoFamilyToDeviceValues_GaugeAndCounter(t *testing.T) {
	body := strings.NewReader(`# TYPE node_network_speed_bytes gauge
node_network_speed_bytes{device="eni0"} 1.25e+09
node_network_speed_bytes{device="lo"} 0
# TYPE node_network_receive_bytes_total counter
node_network_receive_bytes_total{device="eni0"} 5e+08
`)
	families, err := (&expfmt.TextParser{}).TextToMetricFamilies(body)
	require.NoError(t, err)

	speed := dtoFamilyToDeviceValues(families["node_network_speed_bytes"])
	assert.InDelta(t, 1.25e9, speed["eni0"], 1)
	assert.InDelta(t, 0, speed["lo"], 1)

	rx := dtoFamilyToDeviceValues(families["node_network_receive_bytes_total"])
	assert.InDelta(t, 5e8, rx["eni0"], 1)
}

func TestIsIfaceDenylisted(t *testing.T) {
	cases := []struct {
		device   string
		expected bool
	}{
		{"lo", true},
		{"veth123abc", true},
		{"docker0", true},
		{"cni0", true},
		{"cali1234", true},
		{"flannel.1", true},
		{"br-abc123", true},
		{"virbr0", true},
		{"kube-ipvs0", true},
		{"eni0", false},
		{"eni12345", false},
		{"eth0", false},
	}
	for _, tc := range cases {
		t.Run(tc.device, func(t *testing.T) {
			assert.Equal(t, tc.expected, isIfaceDenylisted(tc.device))
		})
	}
}

func TestBandwidthCache_StaleEviction(t *testing.T) {
	c := newBandwidthCache()
	old := networkCounterSample{
		timestamp: time.Now().Add(-2 * counterStaleDuration),
		rxBytes:   100,
	}
	c.store("target", "node", "eni0", old)

	_, ok := c.load("target", "node", "eni0", time.Now())
	assert.False(t, ok, "stale sample should be evicted")

	c.mu.Lock()
	_, exists := c.items[c.cacheKey("target", "node", "eni0")]
	c.mu.Unlock()
	assert.False(t, exists, "stale entry should be deleted from map")
}

func TestBandwidthCache_FreshSample(t *testing.T) {
	c := newBandwidthCache()
	fresh := networkCounterSample{
		timestamp: time.Now().Add(-5 * time.Second),
		rxBytes:   500,
	}
	c.store("target", "node", "eni0", fresh)

	got, ok := c.load("target", "node", "eni0", time.Now())
	require.True(t, ok)
	assert.InDelta(t, fresh.rxBytes, got.rxBytes, 0)
}