package prom

import (
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/skyhook-io/radar/pkg/prom/promtest"
)

// This opt-in suite evaluates the production expressions in real Prometheus,
// rather than teaching a mock server to return the answers the test expects.
func TestWorkloadPromQL(t *testing.T) {
	image := os.Getenv("RADAR_TEST_PROMTOOL_IMAGE")
	if image == "" {
		t.Skip("set RADAR_TEST_PROMTOOL_IMAGE=prom/prometheus:v3.5.0 to evaluate queries with Docker")
	}
	input := []map[string]string{}
	add := func(name, labels, values string) {
		input = append(input, map[string]string{"series": name + "{" + labels + "}", "values": values})
	}
	scope := WorkloadMetricsScope{ClusterLabels: map[string]string{"cluster": "west"}}
	sel := SelectPods("shop", []string{"api-0"})
	tests := []map[string]any{}
	check := func(query, labels string, value float64) {
		tests = append(tests, map[string]any{"expr": "round((" + query + "), 0.000001)", "eval_time": "10m", "exp_samples": []map[string]any{{"labels": labels, "value": value}}})
	}
	identityPods := []WorkloadPodIdentity{{Name: "identity-0", UID: "030a7597-c1fc-48b0-9bb4-683489285358"}, {Name: "identity-1", UID: "0d709b51-f624-4d0a-b7dd-79e17402acb9"}}
	for i, pod := range identityPods {
		base := fmt.Sprintf(`job="custom-http",k8s_namespace_name="shop",k8s_pod_name=%q,k8s_pod_uid=%q,http_response_status_code="200"`, pod.Name, pod.UID)
		add("http_server_request_duration_seconds_count", base, "0+60x10")
		foreign := strings.Replace(base, pod.UID, "ffffffff-ffff-ffff-ffff-ffffffffffff", 1)
		add("http_server_request_duration_seconds_count", foreign, "0+6000x10")
		mismatched := strings.Replace(base, pod.Name, identityPods[1-i].Name, 1) + `,__radar_keep="1",__radar_pair="forged"`
		add("http_server_request_duration_seconds_count", mismatched, "0+6000x10")
		id := "/kubepods/burstable/pod" + pod.UID + "/container"
		if i == 1 {
			id = "/kubepods.slice/kubepods-burstable.slice/kubepods-burstable-pod" + strings.ReplaceAll(pod.UID, "-", "_") + ".slice/container.scope"
		}
		resource := fmt.Sprintf(`namespace="shop",pod=%q,container="app",id=%q`, pod.Name, id)
		add("container_cpu_usage_seconds_total", resource, "0+60x10")
		add("container_cpu_usage_seconds_total", strings.ReplaceAll(resource, pod.Name, identityPods[1-i].Name), "0+6000x10")
		add("container_cpu_usage_seconds_total", fmt.Sprintf(`namespace="shop",pod=%q,container="app",id="/not-a-pod",__radar_uid=%q,__radar_keep="1"`, pod.Name, pod.UID), "0+6000x10")
	}
	identitySelection := SelectPods("shop", []string{"identity-0", "identity-1"})
	identityRequests, err := BuildIdentityRequestQueries(time.Minute, identitySelection, identityPods, RequestSourceBeyla, `job="custom-http"`)
	if err != nil {
		t.Fatal(err)
	}
	check(identityRequests.Rate, "{}", 2)
	check(identityRequests.ObservedPods, "{}", 2)
	identityCPU, err := BuildIdentityWorkloadResourceQuery(time.Minute, identitySelection, identityPods, CategoryCPU)
	if err != nil {
		t.Fatal(err)
	}
	check("sum("+identityCPU+")", "{}", 2)
	kubeletPod := WorkloadPodIdentity{Name: "kubelet-identity", UID: "bf48504c-d21a-4b82-a598-a19868cf3d29"}
	kubeletID := "/kubelet.slice/kubelet-kubepods.slice/kubelet-kubepods-burstable.slice/kubelet-kubepods-burstable-podbf48504c_d21a_4b82_a598_a19868cf3d29.slice/cri-containerd-container.scope"
	kubeletLabels := fmt.Sprintf(`namespace="shop",pod=%q,container="app",id=%q`, kubeletPod.Name, kubeletID)
	for _, fixture := range []struct{ metric, values string }{
		{"container_cpu_usage_seconds_total", "0+60x10"},
		{"container_memory_working_set_bytes", "1024+0x10"},
		{"container_cpu_cfs_throttled_periods_total", "0+15x10"},
		{"container_cpu_cfs_periods_total", "0+60x10"},
	} {
		add(fixture.metric, kubeletLabels, fixture.values)
		add(fixture.metric, strings.ReplaceAll(kubeletLabels, "bf48504c_d21a_4b82_a598_a19868cf3d29", "ffffffff_ffff_ffff_ffff_ffffffffffff"), "0+6000x10")
		add(fixture.metric, strings.ReplaceAll(kubeletLabels, kubeletPod.Name, "wrong-pod-name"), "0+6000x10")
	}
	for _, category := range []MetricCategory{CategoryCPU, CategoryMemory} {
		query, err := BuildIdentityWorkloadResourceQuery(time.Minute, SelectPods("shop", []string{kubeletPod.Name, "wrong-pod-name"}), []WorkloadPodIdentity{kubeletPod}, category)
		if err != nil {
			t.Fatal(err)
		}
		want := float64(1)
		if category == CategoryMemory {
			want = 1024
		}
		check(query, `{pod="kubelet-identity"}`, want)
	}
	kubeletThrottle, err := BuildIdentityThrottleQuery(time.Minute, SelectPods("shop", []string{kubeletPod.Name, "wrong-pod-name"}), []WorkloadPodIdentity{kubeletPod})
	if err != nil {
		t.Fatal(err)
	}
	check(kubeletThrottle, `{pod="kubelet-identity"}`, 25)
	for _, source := range []RequestSource{RequestSourceBeyla, RequestSourceIstio} {
		base := `cluster="west",job="beyla",k8s_namespace_name="shop",k8s_pod_name="api-0"`
		metric, status := "http_server_request_duration_seconds", "http_response_status_code"
		buckets := []string{"0.1", "0.5", "1", "+Inf"}
		if source == RequestSourceIstio {
			base = `cluster="west",reporter="destination",request_protocol="http",namespace="shop",pod="api-0",destination_workload_namespace="shop"`
			metric, status = "istio_request_duration_milliseconds", "response_code"
			buckets = []string{"100", "500", "1000", "+Inf"}
		}
		count := metric + "_count"
		if source == RequestSourceIstio {
			count = "istio_requests_total"
		}
		for i, code := range []string{"200", "500"} {
			increments := []int{108, 12}
			add(count, base+fmt.Sprintf(`,%s="%s"`, status, code), fmt.Sprintf("0+%dx10", increments[i]))
		}
		duplicateBase := base + `,instance="duplicate"`
		for i, code := range []string{"200", "500"} {
			increments := []int{108, 12}
			add(count, duplicateBase+fmt.Sprintf(`,%s="%s"`, status, code), fmt.Sprintf("0+%dx10", increments[i]))
		}
		for i, le := range buckets {
			increments := []int{60, 108, 120, 120}
			for _, labels := range []string{base, duplicateBase} {
				if source == RequestSourceIstio {
					for j, code := range []string{"200", "500"} {
						fraction := []float64{0.9, 0.1}[j]
						add(metric+"_bucket", labels+fmt.Sprintf(`,le=%q,response_code=%q`, le, code), fmt.Sprintf("0+%gx10", float64(increments[i])*fraction))
						if le == "+Inf" {
							add(metric+"_count", labels+fmt.Sprintf(`,response_code=%q`, code), fmt.Sprintf("0+%gx10", 120*fraction))
						}
					}
				} else {
					add(metric+"_bucket", labels+fmt.Sprintf(`,le=%q`, le), fmt.Sprintf("0+%dx10", increments[i]))
				}
			}
		}
		queries, err := BuildRequestQueries(time.Minute, sel, scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(queries.Rate, "{}", 2)
		check(queries.Errors, "{}", 0.2)
		check(queries.P50, "{}", 0.1)
		check(queries.P95, "{}", 0.75)
		if source == RequestSourceIstio {
			check(queries.HistogramCoverage, "{}", 1)
		} else {
			check(queries.HistogramCoverage, "{}", 2)
		}
		check(queries.HistogramUniformity, "{}", 1)
		check(queries.StatusCoverage, "{}", 2)
		check(queries.Population, "{}", 1)
		check(queries.ObservedPods, "{}", 1)
		counterResetBase := strings.ReplaceAll(base, `"api-0"`, `"api-counter-reset"`)
		add(count, counterResetBase+fmt.Sprintf(`,%s="200"`, status), "0+60x6 60+60x123")
		counterReset, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{"api-counter-reset"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(counterReset.Rate, "{}", 1)
		check("count("+counterReset.Errors+") or vector(0)", "{}", 0)
		longWindow, err := BuildRequestQueries(30*time.Minute, SelectPods("shop", []string{"api-counter-reset"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		tests = append(tests, map[string]any{"expr": longWindow.Rate, "eval_time": "120m", "exp_samples": []map[string]any{{"labels": "{}", "value": 1}}})
		partlyObserved, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{"api-0", "api-uninstrumented"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(partlyObserved.ObservedPods, "{}", 1)
		idleBase := strings.ReplaceAll(base, `"api-0"`, `"api-idle"`)
		add(count, idleBase+fmt.Sprintf(`,%s="200"`, status), "0+0x10")
		for _, le := range buckets {
			add(metric+"_bucket", idleBase+fmt.Sprintf(`,le="%s"`, le), "0+0x10")
		}
		idle, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{"api-idle"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(idle.Rate, "{}", 0)
		check(idle.ObservedPods, "{}", 1)
		for _, undefined := range []string{idle.P50, idle.P95} {
			check("("+undefined+") == bool ("+undefined+")", "{}", 0)
		}
		healthyBase := strings.ReplaceAll(base, `"api-0"`, `"api-healthy"`)
		add(count, healthyBase+fmt.Sprintf(`,%s="200"`, status), "0+60x10")
		healthy, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{"api-healthy"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(healthy.Rate, "{}", 1)
		check("count("+healthy.Errors+") or vector(0)", "{}", 0)
		check("count("+healthy.P95+") or vector(0)", "{}", 0)
		check("count("+healthy.HistogramCoverage+") or vector(0)", "{}", 0)
		resetBase := strings.ReplaceAll(base, `"api-0"`, `"api-no-response"`)
		add(count, resetBase+fmt.Sprintf(`,%s="0"`, status), "0+30x10")
		add(count, resetBase+fmt.Sprintf(`,%s="200"`, status), "0+30x10")
		reset, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{"api-no-response"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(reset.Rate, "{}", 1)
		check("count("+reset.Errors+") or vector(0)", "{}", 0)
		check(reset.StatusCoverage, "{}", 1)
		for i, pod := range []string{"api-mixed-1", "api-mixed-2"} {
			mixedBase := strings.ReplaceAll(base, `"api-0"`, `"`+pod+`"`)
			add(count, mixedBase+fmt.Sprintf(`,%s="200"`, status), "0+60x10")
			for _, le := range []string{buckets[i], buckets[2], "+Inf"} {
				add(metric+"_bucket", mixedBase+fmt.Sprintf(`,le="%s"`, le), "0+60x10")
			}
		}
		mixed, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{"api-mixed-1", "api-mixed-2"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(mixed.HistogramUniformity, "{}", 0.5)
		for _, replica := range []string{"a", "b"} {
			haBase := strings.ReplaceAll(base, `"api-0"`, `"api-ha"`) + `,prometheus_replica="` + replica + `"`
			add(count, haBase+fmt.Sprintf(`,%s="200"`, status), "0+60x10")
		}
		ha, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{"api-ha"}), scope, source, "")
		if err != nil {
			t.Fatal(err)
		}
		check(ha.Population, "{}", 2)
	}
	for _, tc := range []struct {
		name                                                         string
		missingStatus, missingBucket, mismatchedCount, foreignBucket bool
	}{
		{name: "async"}, {name: "missing-status", missingStatus: true}, {name: "missing-bucket", missingBucket: true},
		{name: "inconsistent-count", mismatchedCount: true}, {name: "foreign-bucket", foreignBucket: true},
	} {
		pod := "istio-" + tc.name
		base := fmt.Sprintf(`cluster="west",job="istio",reporter="destination",request_protocol="http",namespace="shop",pod=%q,destination_workload_namespace="shop"`, pod)
		for _, code := range []string{"200", "503"} {
			labels := base + fmt.Sprintf(`,response_code=%q`, code)
			add("istio_requests_total", labels, "0+60x10")
			if tc.missingStatus && code == "503" {
				continue
			}
			// The histogram is internally complete but lags its separate counter.
			add("istio_request_duration_milliseconds_count", labels, "0+54x10")
			for _, le := range []string{"100", "500", "+Inf"} {
				if tc.missingBucket && code == "503" && le == "100" {
					continue
				}
				values := "0+54x10"
				if tc.mismatchedCount && code == "503" && le == "+Inf" {
					values = "0+48x10"
				}
				add("istio_request_duration_milliseconds_bucket", labels+fmt.Sprintf(`,le=%q`, le), values)
			}
		}
		if tc.foreignBucket {
			add("istio_request_duration_milliseconds_bucket", base+`,response_code="404",le="100"`, "0+54x10")
		}
		q, err := BuildRequestQueries(time.Minute, SelectPods("shop", []string{pod}), scope, RequestSourceIstio, "")
		if err != nil {
			t.Fatal(err)
		}
		want := float64(0)
		if tc.name == "async" {
			want = 1
			check(q.Rate, "{}", 2)
			check(q.P95, "{}", 0.095)
		}
		check("("+q.HistogramCoverage+") * ("+q.HistogramUniformity+" == bool 1) or vector(0)", "{}", want)
	}
	// Identical workload labels in another cluster must not affect any answer.
	add("http_server_request_duration_seconds_count", `cluster="east",job="beyla",k8s_namespace_name="shop",k8s_pod_name="api-0",http_response_status_code="500"`, "0+9999x10")
	for _, job := range []string{"cadvisor", "duplicate-scrape"} {
		base := fmt.Sprintf(`cluster="west",namespace="shop",pod="api-0",container="app",job="%s"`, job)
		throttled, periods := "0+15x10", "0+60x10"
		if job == "duplicate-scrape" {
			throttled, periods = "0+3x10", "0+30x10"
		}
		add("container_cpu_cfs_throttled_periods_total", base, throttled)
		add("container_cpu_cfs_periods_total", base, periods)
		add("container_cpu_usage_seconds_total", base, "0+60x10")
		add("container_memory_working_set_bytes", base, "1024+0x10")
	}
	query, err := BuildThrottleQuery(time.Minute, sel, scope)
	if err != nil {
		t.Fatal(err)
	}
	check(query, `{pod="api-0"}`, 25)
	for category, want := range map[MetricCategory]float64{CategoryCPU: 1, CategoryMemory: 1024} {
		query, err := BuildWorkloadResourceQuery(time.Minute, sel, scope, category)
		if err != nil {
			t.Fatal(err)
		}
		check(query, `{pod="api-0"}`, want)
	}
	localScope := WorkloadMetricsScope{SingleCluster: true}
	localPods := SelectPods("demo", []string{"web-0"})
	add("http_server_request_duration_seconds_count", `job="app-http-metrics",k8s_namespace_name="demo",k8s_pod_name="web-0",http_response_status_code="200"`, "0+60x10")
	for _, tc := range []struct {
		scope WorkloadMetricsScope
		job   string
		count float64
	}{
		{localScope, "", 0},
		{localScope, `job="app-http-metrics"`, 1},
		{WorkloadMetricsScope{ClusterLabels: map[string]string{"cluster": "skh-nonprod"}}, `job="app-http-metrics"`, 0},
	} {
		queries, err := BuildRequestQueries(time.Minute, localPods, tc.scope, RequestSourceBeyla, tc.job)
		if err != nil {
			t.Fatal(err)
		}
		check("count("+queries.Rate+") or vector(0)", "{}", tc.count)
	}
	add("http_server_request_duration_seconds_count", `job="demo/web",instance="demo.web-0.nginx",http_response_status_code="200"`, "0+60x10")
	add("target_info", `job="demo/web",instance="demo.web-0.nginx",k8s_namespace_name="demo",k8s_pod_name="web-0"`, "1+0x10")
	otlp, err := BuildRequestQueries(time.Minute, localPods, localScope, RequestSourceBeyla, `job="demo/web"`)
	if err != nil {
		t.Fatal(err)
	}
	check("count("+otlp.Rate+") or vector(0)", "{}", 0)
	check(`sum(rate(http_server_request_duration_seconds_count{job="demo/web"}[5m]))`, "{}", 1)
	promtest.Run(t, image, input, tests)
}