package prom

import (
	"context"
	"fmt"
	"reflect"
	"testing"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/client-go/kubernetes/fake"
	ktesting "k8s.io/client-go/testing"
)

func TestScoreService_TableDriven(t *testing.T) {
	tests := []struct {
		name         string
		svc          corev1.Service
		wantMin      int
		wantMax      int
		wantBasePath string
	}{
		{
			name: "plain prometheus by app.kubernetes.io/name + port",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "prometheus-server",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "prometheus"},
				},
				Spec: corev1.ServiceSpec{
					Ports: []corev1.ServicePort{{Port: 9090}},
				},
			},
			wantMin: 100 + 30 + 20 + 10, // name + port + name-contains + metrics ns
			wantMax: 500,
		},
		{
			name: "vmselect sets basePath",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "vmselect",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "vmselect"},
				},
				Spec: corev1.ServiceSpec{
					Ports: []corev1.ServicePort{{Port: 8481}},
				},
			},
			wantMin:      90 + 25 + 20 + 10,
			wantMax:      200,
			wantBasePath: "/select/0/prometheus",
		},
		{
			name: "thanos-query scores lower than prometheus but non-zero",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "thanos-query",
					Namespace: "observability",
					Labels:    map[string]string{"app.kubernetes.io/name": "thanos-query"},
				},
				Spec: corev1.ServiceSpec{
					Ports: []corev1.ServicePort{{Port: 9009}},
				},
			},
			wantMin: 80 + 25 + 15 + 10,
			wantMax: 200,
		},
		{
			name: "unrelated service scores zero",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "redis",
					Namespace: "default",
					Labels:    map[string]string{"app": "redis"},
				},
				Spec: corev1.ServiceSpec{
					Ports: []corev1.ServicePort{{Port: 6379}},
				},
			},
			wantMax: 0,
		},
		{
			name: "ExternalName excluded",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{Name: "prometheus", Namespace: "monitoring"},
				Spec:       corev1.ServiceSpec{Type: corev1.ServiceTypeExternalName},
			},
			wantMax: 0,
		},
		{
			name: "skip-namespace excluded",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "prometheus",
					Namespace: "kube-public",
					Labels:    map[string]string{"app.kubernetes.io/name": "prometheus"},
				},
			},
			wantMax: 0,
		},
		// Non-query components must score zero even though their names contain
		// "prometheus" and they live in a metrics namespace — these are the
		// exact services that were polluting discovery and wasting port-forwards.
		{
			name: "node-exporter excluded despite prometheus in name",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "kube-prometheus-stack-prometheus-node-exporter",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "prometheus-node-exporter"},
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9100}}},
			},
			wantMax: 0,
		},
		{
			name: "kube-state-metrics excluded",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "kube-prometheus-stack-kube-state-metrics",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "kube-state-metrics"},
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 8080}}},
			},
			wantMax: 0,
		},
		{
			name: "prometheus-operator excluded",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "kube-prometheus-stack-operator",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/component": "prometheus-operator"},
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 443}}},
			},
			wantMax: 0,
		},
		{
			name: "alertmanager excluded",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "kube-prometheus-stack-alertmanager",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "alertmanager"},
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9093}}},
			},
			wantMax: 0,
		},
		{
			name: "pushgateway excluded",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "prometheus-pushgateway",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "prometheus-pushgateway"},
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9091}}},
			},
			wantMax: 0,
		},
		{
			name: "definitive prometheus label overrides a deny-list substring in the name",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "grafana-prometheus-server", // "grafana" substring
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "prometheus"},
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9090}}},
			},
			wantMin: 1, // must not be excluded by the "grafana" token
			wantMax: 500,
		},
		{
			name: "prometheus from an operator-named release is kept",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{Name: "prometheus-operator-prometheus", Namespace: "monitoring"},
				Spec:       corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9090}}},
			},
			wantMin: 1, // must not be excluded (name contains "operator" but isn't the operator)
			wantMax: 500,
		},
		{
			name: "kubelet scrape-target excluded despite prometheus in name",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "kube-prometheus-stack-kubelet",
					Namespace: "kube-system",
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 10250}}},
			},
			wantMax: 0,
		},
		{
			name: "thanos-store excluded (not the query API)",
			svc: corev1.Service{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "thanos-store",
					Namespace: "monitoring",
					Labels:    map[string]string{"app.kubernetes.io/name": "thanos-store"},
				},
				Spec: corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 10902}}},
			},
			wantMax: 0,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			score, bp, _ := ScoreService(tc.svc)
			if score < tc.wantMin || (tc.wantMax > 0 && score > tc.wantMax) {
				t.Errorf("score=%d, want in [%d, %d]", score, tc.wantMin, tc.wantMax)
			}
			if tc.wantMax == 0 && score != 0 {
				t.Errorf("score=%d, want 0", score)
			}
			if tc.wantBasePath != "" && bp != tc.wantBasePath {
				t.Errorf("basePath=%q, want %q", bp, tc.wantBasePath)
			}
		})
	}
}

func TestDiscover_WellKnownFirst(t *testing.T) {
	// Install a standard prometheus-server at a well-known location
	// plus an unrelated redis service.
	wellKnown := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "prometheus-server", Namespace: "monitoring"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.1",
			Ports:     []corev1.ServicePort{{Port: 80, TargetPort: intstr.FromInt(9090)}},
		},
	}
	redis := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "redis", Namespace: "default"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.2",
			Ports:     []corev1.ServicePort{{Port: 6379}},
		},
	}
	// Install an additional unknown-but-scoring dynamic candidate.
	thanos := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "thanos-query",
			Namespace: "observability",
			Labels:    map[string]string{"app.kubernetes.io/name": "thanos-query"},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.3",
			Ports:     []corev1.ServicePort{{Port: 9009}},
		},
	}

	k8s := fake.NewSimpleClientset(wellKnown, redis, thanos)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 3})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	if len(cands) < 2 {
		t.Fatalf("want at least 2 candidates, got %d", len(cands))
	}

	// First must be the well-known match.
	if cands[0].Source != CandidateSourceWellKnown {
		t.Errorf("cands[0].Source = %q, want well_known", cands[0].Source)
	}
	if cands[0].Namespace != "monitoring" || cands[0].Name != "prometheus-server" {
		t.Errorf("cands[0] = %s/%s, want monitoring/prometheus-server", cands[0].Namespace, cands[0].Name)
	}
	if cands[0].ClusterAddr != "http://prometheus-server.monitoring.svc.cluster.local:80" {
		t.Errorf("cluster addr = %q", cands[0].ClusterAddr)
	}
	if cands[0].TargetPort != 9090 {
		t.Errorf("TargetPort = %d, want 9090", cands[0].TargetPort)
	}

	// Dynamic thanos match should be present.
	var sawDynamicThanos bool
	for _, c := range cands {
		if c.Source == CandidateSourceDynamic && c.Name == "thanos-query" {
			sawDynamicThanos = true
			break
		}
	}
	if !sawDynamicThanos {
		t.Errorf("expected dynamic thanos candidate; got %+v", cands)
	}

	// Redis must not appear in any form.
	for _, c := range cands {
		if c.Name == "redis" {
			t.Errorf("redis should not be a candidate: %+v", c)
		}
	}
}

func TestDiscover_DeDupesWellKnownAndDynamic(t *testing.T) {
	// A service that matches both a well-known location (monitoring/prometheus-server)
	// and scores in the dynamic scan (app.kubernetes.io/name=prometheus) must
	// appear exactly once, sourced from the well-known layer.
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "prometheus-server",
			Namespace: "monitoring",
			Labels:    map[string]string{"app.kubernetes.io/name": "prometheus"},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.1",
			Ports:     []corev1.ServicePort{{Port: 9090}},
		},
	}

	k8s := fake.NewSimpleClientset(svc)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}

	var count int
	for _, c := range cands {
		if c.Namespace == "monitoring" && c.Name == "prometheus-server" {
			count++
			if c.Source != CandidateSourceWellKnown {
				t.Errorf("de-duped candidate should retain well_known source, got %q", c.Source)
			}
		}
	}
	if count != 1 {
		t.Fatalf("want monitoring/prometheus-server exactly once, got %d (all: %+v)", count, cands)
	}
}

func TestDiscover_ExcludesControlPlaneScrapeTargets(t *testing.T) {
	// A kube-prometheus-stack control-plane scrape-target Service carries
	// "prometheus" in its name but only proxies a component's /metrics; it must
	// not become a candidate — it can't answer PromQL and would waste a
	// port-forward. Excluded by name (a component token), not by score.
	noise := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "kube-prometheus-stack-kube-etcd",
			Namespace: "kube-system",
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "None",
			Ports:     []corev1.ServicePort{{Port: 2381}},
		},
	}

	k8s := fake.NewSimpleClientset(noise)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	if len(cands) != 0 {
		t.Fatalf("control-plane scrape-target service became a candidate: %+v", cands)
	}
}

func TestDiscover_ExcludesNamespaceOnlyNeighbor(t *testing.T) {
	// An unrelated Service that only matches by sharing a metrics namespace (no
	// name/label/port Prometheus signal) must NOT become a candidate — discovery
	// probes candidates with the operator's configured Prometheus headers, so a
	// namespace neighbor would receive credentials it shouldn't.
	neighbor := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "billing-api",
			Namespace: "monitoring",
			Labels:    map[string]string{"app": "billing"},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.42",
			Ports:     []corev1.ServicePort{{Port: 8080, Name: "http"}},
		},
	}

	if _, _, identity := ScoreService(*neighbor); identity {
		t.Fatal("namespace-only neighbor reported a Prometheus identity signal")
	}

	k8s := fake.NewSimpleClientset(neighbor)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	if len(cands) != 0 {
		t.Fatalf("namespace-only neighbor became a candidate (would be probed with creds): %+v", cands)
	}
}

func TestDiscover_KeepsPrometheusWithNamedTargetPort(t *testing.T) {
	// An unlabeled Prometheus behind service port 80 with a *named* targetPort
	// scores only on its name (a named targetPort has no numeric value to
	// credit). Score ranks, it does not gate — so this real endpoint must still
	// be discovered and left for the probe to validate, not silently dropped.
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "prometheus-server", Namespace: "apps"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.6",
			Ports:     []corev1.ServicePort{{Port: 80, TargetPort: intstr.FromString("web")}},
		},
	}

	k8s := fake.NewSimpleClientset(svc)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	var found bool
	for _, c := range cands {
		if c.Namespace == "apps" && c.Name == "prometheus-server" {
			found = true
		}
	}
	if !found {
		t.Fatalf("prometheus with a named targetPort was pruned: %+v", cands)
	}
}

func TestDiscover_KeepsUnlabeledThanosQuery(t *testing.T) {
	// An unlabeled thanos-query on its common HTTP port is a real query API and
	// must survive the dynamic score floor (name match + recognized port).
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "thanos-query", Namespace: "obs"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.8",
			Ports:     []corev1.ServicePort{{Port: 10902}},
		},
	}

	k8s := fake.NewSimpleClientset(svc)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	var found bool
	for _, c := range cands {
		if c.Name == "thanos-query" {
			found = true
		}
	}
	if !found {
		t.Fatalf("unlabeled thanos-query was pruned: %+v", cands)
	}
}

func TestDiscover_KeepsUnlabeledCustomPrometheus(t *testing.T) {
	// A real Prometheus in a non-standard namespace with no conventional labels
	// and a service port of 80 scores only on the name match, but it is a
	// genuine query endpoint and must still be discovered — pruning must not be
	// so aggressive that it drops it.
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "prometheus-server", Namespace: "platform"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.9",
			Ports:     []corev1.ServicePort{{Port: 80, TargetPort: intstr.FromInt(9090)}},
		},
	}

	k8s := fake.NewSimpleClientset(svc)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	var found bool
	for _, c := range cands {
		if c.Namespace == "platform" && c.Name == "prometheus-server" {
			found = true
		}
	}
	if !found {
		t.Fatalf("unlabeled custom prometheus was pruned: %+v", cands)
	}
}

func TestDiscover_SkipsDynamicWhenDisabled(t *testing.T) {
	// Only a dynamic-scoring service is present (no well-known match).
	prom := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "my-prometheus",
			Namespace: "observability",
			Labels:    map[string]string{"app.kubernetes.io/name": "prometheus"},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.5",
			Ports:     []corev1.ServicePort{{Port: 9090}},
		},
	}

	k8s := fake.NewSimpleClientset(prom)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: false})
	if err != nil {
		t.Fatal(err)
	}
	if len(cands) != 0 {
		t.Errorf("expected no candidates when dynamic is disabled and no well-known match; got %d", len(cands))
	}
}

func TestDiscover_HeadlessServiceUsesPlainServiceDNS(t *testing.T) {
	headless := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "prometheus-server", Namespace: "monitoring"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "None",
			Ports:     []corev1.ServicePort{{Port: 9090}},
		},
	}
	k8s := fake.NewSimpleClientset(headless)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{})
	if err != nil {
		t.Fatal(err)
	}
	if len(cands) != 1 {
		t.Fatalf("want 1 candidate, got %d", len(cands))
	}
	// A headless Service publishes A records for its ready pods under its own
	// name; the old {service}-0. pod prefix assumed the StatefulSet was named
	// after the Service and left non-conforming backends unreachable.
	want := "http://prometheus-server.monitoring.svc.cluster.local:9090"
	if cands[0].ClusterAddr != want {
		t.Errorf("cluster addr = %q, want %q", cands[0].ClusterAddr, want)
	}
}

func TestDiscover_NilClient(t *testing.T) {
	_, err := Discover(context.Background(), nil, DiscoverOptions{})
	if err == nil {
		t.Error("expected error for nil client")
	}
}

// TestDiscover_LoggerCalledSerially guards the DiscoverOptions.Logger contract:
// well-known lookups run concurrently, but the Logger must not be — a caller can
// pass an unsynchronized callback. The reactor makes every well-known Get fail
// (non-NotFound) so each would log; an unsynchronized slice append from parallel
// workers would trip the race detector.
func TestDiscover_LoggerCalledSerially(t *testing.T) {
	k8s := fake.NewSimpleClientset()
	k8s.PrependReactor("get", "services", func(ktesting.Action) (bool, runtime.Object, error) {
		return true, nil, fmt.Errorf("simulated API error")
	})

	var logs []string
	_, err := Discover(context.Background(), k8s, DiscoverOptions{
		Logger: func(format string, args ...interface{}) {
			logs = append(logs, fmt.Sprintf(format, args...)) // deliberately unsynchronized
		},
	})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	if len(logs) == 0 {
		t.Fatal("expected error logs from failed well-known lookups")
	}
}

// TestDiscover_CarettaVMRanksAfterDynamicPrometheus pins that caretta-vm, a
// last-resort fallback that commonly lacks workload metrics, never outranks a
// real Prometheus — including one discoverable only via the dynamic scan
// (non-standard namespace/name). Before the fallback tier existed, caretta-vm
// (a well-known location) sorted ahead of every dynamic candidate.
func TestDiscover_CarettaVMRanksAfterDynamicPrometheus(t *testing.T) {
	carettaVM := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "caretta-vm", Namespace: "caretta"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.9",
			Ports:     []corev1.ServicePort{{Port: 8428}},
		},
	}
	// A real Prometheus at a non-standard location — found only by the dynamic
	// scan via its app.kubernetes.io/name identity label.
	dynamicProm := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "prom",
			Namespace: "myteam",
			Labels:    map[string]string{"app.kubernetes.io/name": "prometheus"},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.10",
			Ports:     []corev1.ServicePort{{Port: 9090}},
		},
	}

	k8s := fake.NewSimpleClientset(carettaVM, dynamicProm)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}

	idx := func(ns, name string) int {
		for i, c := range cands {
			if c.Namespace == ns && c.Name == name {
				return i
			}
		}
		return -1
	}
	dynIdx := idx("myteam", "prom")
	carettaIdx := idx("caretta", "caretta-vm")
	if dynIdx < 0 {
		t.Fatalf("dynamic prometheus not discovered: %+v", cands)
	}
	if carettaIdx < 0 {
		t.Fatalf("caretta-vm fallback not discovered: %+v", cands)
	}
	if carettaIdx < dynIdx {
		t.Fatalf("caretta-vm (idx %d) ranked before dynamic prometheus (idx %d) — fallback must rank last", carettaIdx, dynIdx)
	}
	if carettaIdx != len(cands)-1 {
		t.Fatalf("caretta-vm at idx %d, want last candidate (%d): %+v", carettaIdx, len(cands)-1, cands)
	}
}

// TestDiscover_CarettaVMReturnedWhenSoleBackend guards the safety net: when
// caretta-vm is the only metrics backend, it must still be returned so the
// cluster keeps working — demoting it to a fallback tier must not drop it.
func TestDiscover_CarettaVMReturnedWhenSoleBackend(t *testing.T) {
	carettaVM := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "caretta-vm", Namespace: "caretta"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.9",
			Ports:     []corev1.ServicePort{{Port: 8428}},
		},
	}
	k8s := fake.NewSimpleClientset(carettaVM)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	if len(cands) != 1 || cands[0].Namespace != "caretta" || cands[0].Name != "caretta-vm" {
		t.Fatalf("want sole caretta-vm candidate, got %+v", cands)
	}
}

// TestDiscover_FallbackSurvivesListFailure guards finding-1: when the cluster
// list fails (get-but-not-list RBAC) but caretta-vm is fetchable via targeted
// Get, the fallback tier must still be returned — not silently dropped.
func TestDiscover_FallbackSurvivesListFailure(t *testing.T) {
	carettaVM := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "caretta-vm", Namespace: "caretta"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.9",
			Ports:     []corev1.ServicePort{{Port: 8428}},
		},
	}
	k8s := fake.NewSimpleClientset(carettaVM)
	k8s.PrependReactor("list", "services", func(ktesting.Action) (bool, runtime.Object, error) {
		return true, nil, fmt.Errorf("forbidden: cannot list services at cluster scope")
	})
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	if len(cands) != 1 || cands[0].Namespace != "caretta" || cands[0].Name != "caretta-vm" {
		t.Fatalf("fallback dropped on list failure: got %+v", cands)
	}
}

// TestDiscover_CarettaVMLastEvenWithNonStandardServicePort guards finding-2: a
// caretta-vm Service fronted by service port 80 -> container 8428 keys as :80 in
// the dynamic scan but :8428 in the fallback tier. A port-keyed dedup alone would
// let it reappear among dynamics and lose the last-place invariant; name-based
// suppression keeps it single and last.
func TestDiscover_CarettaVMLastEvenWithNonStandardServicePort(t *testing.T) {
	carettaVM := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "caretta-vm", Namespace: "caretta"},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.9",
			Ports:     []corev1.ServicePort{{Port: 80, TargetPort: intstr.FromInt(8428)}},
		},
	}
	dynamicProm := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "prom",
			Namespace: "myteam",
			Labels:    map[string]string{"app.kubernetes.io/name": "prometheus"},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.10",
			Ports:     []corev1.ServicePort{{Port: 9090}},
		},
	}
	k8s := fake.NewSimpleClientset(carettaVM, dynamicProm)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	carettaCount, carettaIdx := 0, -1
	for i, c := range cands {
		if c.Name == "caretta-vm" {
			carettaCount++
			carettaIdx = i
		}
	}
	if carettaCount != 1 {
		t.Fatalf("caretta-vm appeared %d times, want 1: %+v", carettaCount, cands)
	}
	if carettaIdx != len(cands)-1 {
		t.Fatalf("caretta-vm at idx %d, want last (%d): %+v", carettaIdx, len(cands)-1, cands)
	}
}

// TestDiscover_ExcludesPortOnlyPortMatch reproduces issue #1356: a Service that
// matches only on a well-known port NUMBER (9090 here, shared with gRPC plugins
// such as CloudNativePG's barman-cloud) carries no Prometheus-family identity —
// no name, label, or port-name signal — and must NOT be admitted as a candidate.
// Admitting it wastes a doomed port-forward that, in-cluster on a read-only
// ServiceAccount, surfaces to the user as a confusing "Prometheus not connected".
func TestDiscover_ExcludesPortOnlyPortMatch(t *testing.T) {
	barman := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:        "barman-cloud",
			Namespace:   "cnpg-system",
			Annotations: map[string]string{"cnpg.io/pluginPort": "9090"},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "10.0.0.20",
			Ports:     []corev1.ServicePort{{Port: 9090}},
		},
	}

	// Ranking signal is present (port scores) but identity must be false.
	score, _, identity := ScoreService(*barman)
	if identity {
		t.Fatalf("barman-cloud identity=true (a bare port number must not admit); score=%d", score)
	}
	if score == 0 {
		t.Fatalf("expected a non-zero ranking score from the port number, got 0")
	}

	k8s := fake.NewSimpleClientset(barman)
	cands, err := Discover(context.Background(), k8s, DiscoverOptions{IncludeDynamic: true, MaxDynamic: 5})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	for _, c := range cands {
		if c.Name == "barman-cloud" {
			t.Fatalf("port-only gRPC service admitted as a Prometheus candidate: %+v", cands)
		}
	}
}

// TestScoreService_IdentityRequiresFamilySignal locks the issue #1356 contract at
// the scoring layer: identity is set by a name/label/port-name signal, never by a
// bare port number.
func TestScoreService_IdentityRequiresFamilySignal(t *testing.T) {
	cases := []struct {
		name      string
		svc       corev1.Service
		wantIdent bool
	}{
		{"port-number only", corev1.Service{
			ObjectMeta: metav1.ObjectMeta{Name: "some-grpc-svc", Namespace: "apps"},
			Spec:       corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9090}}},
		}, false},
		{"metrics namespace + port only", corev1.Service{
			ObjectMeta: metav1.ObjectMeta{Name: "some-grpc-svc", Namespace: "monitoring"},
			Spec:       corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 8428}}},
		}, false},
		{"label signal admits", corev1.Service{
			ObjectMeta: metav1.ObjectMeta{Name: "x", Namespace: "apps", Labels: map[string]string{"app.kubernetes.io/name": "prometheus"}},
			Spec:       corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9090}}},
		}, true},
		{"name signal admits", corev1.Service{
			ObjectMeta: metav1.ObjectMeta{Name: "team-prometheus", Namespace: "apps"},
			Spec:       corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9090}}},
		}, true},
		{"port-name signal admits", corev1.Service{
			ObjectMeta: metav1.ObjectMeta{Name: "custom", Namespace: "apps"},
			Spec:       corev1.ServiceSpec{Ports: []corev1.ServicePort{{Port: 9090, Name: "prometheus"}}},
		}, true},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			_, _, identity := ScoreService(tc.svc)
			if identity != tc.wantIdent {
				t.Fatalf("identity=%v, want %v", identity, tc.wantIdent)
			}
		})
	}
}

// A headless Service publishes A records for its ready pods under its own name.
// Addressing it as {service}-0.{service} assumes the StatefulSet is named after
// the Service — true for caretta-vm, false for kube-prometheus-stack's
// prometheus-operated, whose pod is
// prometheus-{release}-kube-prometheus-stack-prometheus-0 — and made a reachable
// backend look unreachable in-cluster.
func TestBuildClusterAddrUsesPlainServiceDNS(t *testing.T) {
	tests := []struct {
		name, namespace string
		port            int
		want            string
	}{
		// Headless kube-prometheus-stack Service: its pod is not named
		// prometheus-operated-0, so the old {service}-0. prefix never resolved.
		{"prometheus-operated", "monitoring", 9090, "http://prometheus-operated.monitoring.svc.cluster.local:9090"},
		// Headless caretta store: must still resolve under its own name.
		{"caretta-vm", "default", 8428, "http://caretta-vm.default.svc.cluster.local:8428"},
		// Regular ClusterIP Service: unchanged.
		{"kube-prometheus-stack-prometheus", "monitoring", 9090, "http://kube-prometheus-stack-prometheus.monitoring.svc.cluster.local:9090"},
	}
	for _, tc := range tests {
		if got := buildClusterAddr(tc.name, tc.namespace, tc.port); got != tc.want {
			t.Errorf("buildClusterAddr(%s/%s) = %q, want %q", tc.namespace, tc.name, got, tc.want)
		}
	}
}

// discoverFixture is a cluster with every candidate tier: a well-known
// Prometheus, the caretta-vm fallback, more equal-score dynamic candidates
// than MaxDynamic keeps, and unrelated Services.
func discoverFixture() []*corev1.Service {
	svc := func(ns, name string, lbls map[string]string, port int32) *corev1.Service {
		return &corev1.Service{
			ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns, Labels: lbls},
			Spec:       corev1.ServiceSpec{ClusterIP: "10.0.0.1", Ports: []corev1.ServicePort{{Port: port}}},
		}
	}
	promLabels := map[string]string{"app.kubernetes.io/name": "prometheus"}
	out := []*corev1.Service{
		svc("monitoring", "prometheus-server", nil, 9090),
		svc("caretta", "caretta-vm", nil, 8428),
		svc("default", "redis", nil, 6379),
		svc("kube-system", "kube-dns", nil, 53),
	}
	for _, ns := range []string{"team-g", "team-c", "team-a", "team-f", "team-e", "team-b", "team-d"} {
		out = append(out, svc(ns, "prom", promLabels, 9090))
	}
	return out
}

// TestDiscover_ServicesSourceMatchesAPI pins that enumerating from a cached
// Service set returns exactly what the API path returns, whatever order the
// cache hands Services back in, and makes no API calls.
func TestDiscover_ServicesSourceMatchesAPI(t *testing.T) {
	fixture := discoverFixture()
	objs := make([]runtime.Object, len(fixture))
	for i, s := range fixture {
		objs[i] = s
	}
	want, err := Discover(context.Background(), fake.NewSimpleClientset(objs...), DiscoverOptions{IncludeDynamic: true})
	if err != nil {
		t.Fatalf("API Discover: %v", err)
	}
	if len(want) != 7 {
		t.Fatalf("fixture should yield 1 well-known + 5 dynamic + caretta, got %d: %+v", len(want), want)
	}

	for _, order := range [][]int{{10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0}, {3, 7, 0, 9, 1, 10, 5, 2, 8, 4, 6}} {
		shuffled := make([]*corev1.Service, len(order))
		for i, j := range order {
			shuffled[i] = fixture[j]
		}
		api := fake.NewSimpleClientset()
		got, err := Discover(context.Background(), api, DiscoverOptions{
			IncludeDynamic: true,
			Services:       func() ([]*corev1.Service, bool) { return shuffled, true },
		})
		if err != nil {
			t.Fatalf("cached Discover: %v", err)
		}
		if !reflect.DeepEqual(got, want) {
			t.Fatalf("cached enumeration differs from API enumeration\n got: %+v\nwant: %+v", got, want)
		}
		if actions := api.Actions(); len(actions) != 0 {
			t.Fatalf("cached enumeration made %d API calls: %v", len(actions), actions)
		}
	}
}

// TestDiscover_ServicesSourceDeclinedUsesAPI pins the fallback: a source that
// cannot vouch for a complete view (ok=false) leaves enumeration to the API.
func TestDiscover_ServicesSourceDeclinedUsesAPI(t *testing.T) {
	api := fake.NewSimpleClientset(discoverFixture()[0])
	got, err := Discover(context.Background(), api, DiscoverOptions{
		IncludeDynamic: true,
		Services:       func() ([]*corev1.Service, bool) { return nil, false },
	})
	if err != nil {
		t.Fatalf("Discover: %v", err)
	}
	if len(got) != 1 || got[0].Name != "prometheus-server" {
		t.Fatalf("want the API's monitoring/prometheus-server, got %+v", got)
	}
	if len(api.Actions()) == 0 {
		t.Fatal("declined source should fall back to API calls")
	}
}

func TestIsCandidateService(t *testing.T) {
	fixture := discoverFixture()
	for _, tc := range []struct {
		svc  *corev1.Service
		want bool
	}{
		{fixture[0], true}, // well-known location, no labels
		{fixture[1], true}, // caretta-vm fallback location
		{fixture[2], false},
		{fixture[3], false},
		{fixture[4], true}, // identity label
		{nil, false},
	} {
		name := "nil"
		if tc.svc != nil {
			name = tc.svc.Namespace + "/" + tc.svc.Name
		}
		if got := IsCandidateService(tc.svc); got != tc.want {
			t.Errorf("IsCandidateService(%s) = %v, want %v", name, got, tc.want)
		}
	}
}