// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package api

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
	fodcv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/fodc/v1"
	"github.com/apache/skywalking-banyandb/fodc/proxy/internal/cluster"
	"github.com/apache/skywalking-banyandb/fodc/proxy/internal/lifecycle"
	"github.com/apache/skywalking-banyandb/fodc/proxy/internal/metrics"
	"github.com/apache/skywalking-banyandb/fodc/proxy/internal/registry"
	"github.com/apache/skywalking-banyandb/pkg/logger"
)

func initTestLogger(t *testing.T) {
	t.Helper()
	initErr := logger.Init(logger.Logging{Env: "dev", Level: "debug"})
	require.NoError(t, initErr)
}

// mockRequestSender is a mock implementation of metrics.RequestSender for testing.
type mockRequestSender struct {
	requestErr error
}

func (m *mockRequestSender) RequestMetrics(_ string, _, _ *time.Time) error {
	return m.requestErr
}

// mockClusterDataRequester is a mock implementation of cluster.RequestSender for testing.
type mockClusterDataRequester struct {
	requestErr error
	topologies map[string]*fodcv1.Topology
	clusterMgr *cluster.Manager
}

func (m *mockClusterDataRequester) RequestClusterData(agentID string) error {
	if m.requestErr != nil {
		return m.requestErr
	}
	if topology, exists := m.topologies[agentID]; exists && m.clusterMgr != nil {
		m.clusterMgr.UpdateClusterTopology(agentID, topology)
	}
	return nil
}

func (m *mockClusterDataRequester) setTopology(agentID string, topology *fodcv1.Topology) {
	if m.topologies == nil {
		m.topologies = make(map[string]*fodcv1.Topology)
	}
	m.topologies[agentID] = topology
}

func newTestServer(t *testing.T) (*Server, *registry.AgentRegistry, *mockClusterDataRequester) {
	t.Helper()
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)
	mockRequester := &mockClusterDataRequester{}
	clusterMgr := cluster.NewManager(testRegistry, mockRequester, testLogger)
	mockRequester.clusterMgr = clusterMgr
	server := NewServer(aggregator, clusterMgr, nil, testRegistry, nil, nil, nil, testLogger)
	return server, testRegistry, mockRequester
}

func newTestServerOnly(t *testing.T) *Server {
	t.Helper()
	server, testRegistry, mockRequester := newTestServer(t)
	_ = testRegistry
	_ = mockRequester
	return server
}

func createTestMetric(name string, value float64, labels map[string]string, timestamp time.Time) *metrics.AggregatedMetric {
	return createTestMetricWithType(name, value, labels, timestamp, "")
}

// createTestMetricWithType creates a test metric with an explicit Prometheus type.
func createTestMetricWithType(name string, value float64, labels map[string]string, timestamp time.Time, metricType string) *metrics.AggregatedMetric {
	if labels == nil {
		labels = make(map[string]string)
	}
	return &metrics.AggregatedMetric{
		Name:        name,
		Value:       value,
		Labels:      labels,
		Timestamp:   timestamp,
		AgentID:     "test-agent-1",
		Description: "Test metric description",
		Type:        metricType,
	}
}

func TestHandleClusterTopology_Success(t *testing.T) {
	server, testRegistry, mockRequester := newTestServer(t)
	ctx := context.Background()
	identity := registry.AgentIdentity{
		PodName:        "test-pod",
		Role:           "datanode",
		ContainerNames: []string{"banyandb"},
	}
	agentID, registerErr := testRegistry.RegisterAgent(ctx, identity)
	require.NoError(t, registerErr)
	topology := &fodcv1.Topology{
		Nodes: []*databasev1.Node{
			{
				Metadata: &commonv1.Metadata{
					Name: "node1",
				},
			},
			{
				Metadata: &commonv1.Metadata{
					Name: "test-node",
				},
			},
		},
		Calls: []*fodcv1.Call{
			{
				Id:     "call-1",
				Target: "test-node",
				Source: "node1",
			},
		},
	}
	mockRequester.setTopology(agentID, topology)
	req := httptest.NewRequest(http.MethodGet, "/cluster/topology", nil)
	resp := httptest.NewRecorder()
	server.handleClusterTopology(resp, req)
	assert.Equal(t, http.StatusOK, resp.Code)
	var result fodcv1.Topology
	decodeErr := json.NewDecoder(resp.Body).Decode(&result)
	require.NoError(t, decodeErr)
	assert.Equal(t, 2, len(result.Nodes))
	assert.Equal(t, 1, len(result.Calls))
	assert.Equal(t, "call-1", result.Calls[0].Id)
}

func TestHandleClusterTopology_NoClusterStateCollector(t *testing.T) {
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)
	server := NewServer(aggregator, nil, nil, testRegistry, nil, nil, nil, testLogger)
	req := httptest.NewRequest(http.MethodGet, "/cluster/topology", nil)
	resp := httptest.NewRecorder()
	server.handleClusterTopology(resp, req)
	assert.Equal(t, http.StatusServiceUnavailable, resp.Code)
}

func TestHandleClusterTopology_RequestClusterDataError(t *testing.T) {
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)
	mockRequester := &mockClusterDataRequester{requestErr: assert.AnError}
	clusterMgr := cluster.NewManager(testRegistry, mockRequester, testLogger)
	server := NewServer(aggregator, clusterMgr, nil, testRegistry, nil, nil, nil, testLogger)
	ctx := context.Background()
	identity := registry.AgentIdentity{
		PodName:        "test-pod",
		Role:           "datanode",
		ContainerNames: []string{"banyandb"},
	}
	_, registerErr := testRegistry.RegisterAgent(ctx, identity)
	require.NoError(t, registerErr)
	req := httptest.NewRequest(http.MethodGet, "/cluster/topology", nil)
	resp := httptest.NewRecorder()
	server.handleClusterTopology(resp, req)
	assert.Equal(t, http.StatusOK, resp.Code)
}

func TestNewServer(t *testing.T) {
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)

	server := NewServer(aggregator, nil, nil, testRegistry, nil, nil, nil, testLogger)

	assert.NotNil(t, server)
	assert.Equal(t, aggregator, server.metricsAggregator)
	assert.Equal(t, testRegistry, server.registry)
	assert.WithinDuration(t, time.Now(), server.startTime, time.Second)
}

func TestServer_StartStop(t *testing.T) {
	server := newTestServerOnly(t)

	err := server.Start("localhost:0", 5*time.Second, 5*time.Second)
	require.NoError(t, err)
	assert.NotNil(t, server.server)

	time.Sleep(50 * time.Millisecond)

	stopErr := server.Stop()
	assert.NoError(t, stopErr)
}

func TestServer_Stop_NotStarted(t *testing.T) {
	server := newTestServerOnly(t)

	err := server.Stop()
	assert.NoError(t, err)
}

func TestHandleMetrics_Success(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
	w := httptest.NewRecorder()

	server.handleMetrics(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
	assert.Contains(t, w.Header().Get("Content-Type"), "text/plain")
}

func TestHandleMetrics_MethodNotAllowed(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodPost, "/metrics", nil)
	w := httptest.NewRecorder()

	server.handleMetrics(w, req)

	assert.Equal(t, http.StatusMethodNotAllowed, w.Code)
}

func TestHandleMetrics_WithRoleFilter(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodGet, "/metrics?role=datanode-cold", nil)
	w := httptest.NewRecorder()

	server.handleMetrics(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
}

func TestHandleMetrics_WithAddressFilter(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodGet, "/metrics?address=192.168.1.1", nil)
	w := httptest.NewRecorder()

	server.handleMetrics(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
}

func TestHandleMetrics_CollectionError(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	req = req.WithContext(ctx)
	w := httptest.NewRecorder()

	server.handleMetrics(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
}

func TestHandleMetricsWindows_Success(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	startTime := now.Add(-1 * time.Hour).Format(time.RFC3339)
	endTime := now.Format(time.RFC3339)

	u, urlErr := url.Parse("/metrics-windows")
	require.NoError(t, urlErr)
	q := u.Query()
	q.Set("start_time", startTime)
	q.Set("end_time", endTime)
	u.RawQuery = q.Encode()

	req := httptest.NewRequest(http.MethodGet, u.String(), nil)
	w := httptest.NewRecorder()

	server.handleMetricsWindows(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
	assert.Equal(t, "application/json", w.Header().Get("Content-Type"))

	var response []map[string]interface{}
	decodeErr := json.NewDecoder(w.Body).Decode(&response)
	require.NoError(t, decodeErr)
}

func TestHandleMetricsWindows_InvalidStartTime(t *testing.T) {
	server := newTestServerOnly(t)

	u, urlErr := url.Parse("/metrics-windows")
	require.NoError(t, urlErr)
	q := u.Query()
	q.Set("start_time", "invalid")
	u.RawQuery = q.Encode()

	req := httptest.NewRequest(http.MethodGet, u.String(), nil)
	w := httptest.NewRecorder()

	server.handleMetricsWindows(w, req)

	assert.Equal(t, http.StatusBadRequest, w.Code)
	assert.Contains(t, w.Body.String(), "Invalid start_time format")
}

func TestHandleMetricsWindows_InvalidEndTime(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	startTime := now.Add(-1 * time.Hour).Format(time.RFC3339)

	u, urlErr := url.Parse("/metrics-windows")
	require.NoError(t, urlErr)
	q := u.Query()
	q.Set("start_time", startTime)
	q.Set("end_time", "invalid")
	u.RawQuery = q.Encode()

	req := httptest.NewRequest(http.MethodGet, u.String(), nil)
	w := httptest.NewRecorder()

	server.handleMetricsWindows(w, req)

	assert.Equal(t, http.StatusBadRequest, w.Code)
	assert.Contains(t, w.Body.String(), "Invalid end_time format")
}

func TestHandleMetricsWindows_MethodNotAllowed(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodPost, "/metrics-windows", nil)
	w := httptest.NewRecorder()

	server.handleMetricsWindows(w, req)

	assert.Equal(t, http.StatusMethodNotAllowed, w.Code)
}

func TestHandleMetricsWindows_WithoutTimeWindow(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodGet, "/metrics-windows", nil)
	w := httptest.NewRecorder()

	server.handleMetricsWindows(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
	var response []map[string]interface{}
	decodeErr := json.NewDecoder(w.Body).Decode(&response)
	require.NoError(t, decodeErr)
}

func TestHandleHealth_Success(t *testing.T) {
	server, testRegistry, _ := newTestServer(t)

	ctx := context.Background()
	identity1 := registry.AgentIdentity{PodName: "192.168.1.1", Role: "datanode-cold", ContainerNames: []string{"data"}}
	identity2 := registry.AgentIdentity{PodName: "192.168.1.2", Role: "liaison", ContainerNames: []string{"liaison"}}

	_, err1 := testRegistry.RegisterAgent(ctx, identity1)
	require.NoError(t, err1)
	_, err2 := testRegistry.RegisterAgent(ctx, identity2)
	require.NoError(t, err2)

	req := httptest.NewRequest(http.MethodGet, "/health", nil)
	w := httptest.NewRecorder()

	server.handleHealth(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
	assert.Equal(t, "application/json", w.Header().Get("Content-Type"))

	var response map[string]interface{}
	decodeErr := json.NewDecoder(w.Body).Decode(&response)
	require.NoError(t, decodeErr)
	assert.Equal(t, "healthy", response["status"])
	assert.Equal(t, float64(2), response["agents_total"])
	assert.GreaterOrEqual(t, response["uptime_seconds"], float64(0))
}

func TestHandleHealth_MethodNotAllowed(t *testing.T) {
	server := newTestServerOnly(t)

	req := httptest.NewRequest(http.MethodPost, "/health", nil)
	w := httptest.NewRecorder()

	server.handleHealth(w, req)

	assert.Equal(t, http.StatusMethodNotAllowed, w.Code)
}

func TestFormatPrometheusText_Empty(t *testing.T) {
	server := newTestServerOnly(t)

	result := server.formatPrometheusText([]*metrics.AggregatedMetric{})

	assert.Empty(t, result)
}

func TestFormatPrometheusText_SingleMetric(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Contains(t, result, "# HELP cpu_usage")
	assert.Contains(t, result, "# TYPE cpu_usage gauge")
	assert.Contains(t, result, "cpu_usage")
	assert.Contains(t, result, "75.5")
}

func TestFormatPrometheusText_MultipleMetrics(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
		createTestMetric("cpu_usage", 80.0, map[string]string{"cpu": "1", "pod_name": "192.168.1.1"}, now),
		createTestMetric("memory_usage", 50.0, map[string]string{"type": "heap", "pod_name": "192.168.1.1"}, now),
	}

	result := server.formatPrometheusText(metricsList)

	assert.NotEmpty(t, result)
	assert.Contains(t, result, "cpu_usage")
	assert.Contains(t, result, "memory_usage")
}

func TestFormatPrometheusText_WithLabels(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "env": "test", "pod_name": "192.168.1.1"}, now),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Contains(t, result, "cpu_usage{")
	assert.Contains(t, result, "cpu=\"0\"")
	assert.Contains(t, result, "env=\"test\"")
}

func TestFormatPrometheusText_Histogram(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("request_latency_bucket", 10, map[string]string{"le": "0.1", "ip": "192.168.1.1", "port": "8080"}, now),
		createTestMetric("request_latency_bucket", 25, map[string]string{"le": "0.5", "ip": "192.168.1.1", "port": "8080"}, now),
		createTestMetric("request_latency_bucket", 30, map[string]string{"le": "+Inf", "ip": "192.168.1.1", "port": "8080"}, now),
		createTestMetric("request_latency_sum", 12.5, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
		createTestMetric("request_latency_count", 30, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Contains(t, result, "# TYPE request_latency histogram")
	assert.Contains(t, result, "request_latency_bucket{")
	assert.Contains(t, result, "request_latency_sum{")
	assert.Contains(t, result, "request_latency_count{")
	assert.Contains(t, result, "le=\"0.1\"")
	assert.Contains(t, result, "le=\"0.5\"")
	assert.Contains(t, result, "le=\"+Inf\"")
	assert.NotContains(t, result, "# TYPE request_latency_bucket")
	assert.NotContains(t, result, "# TYPE request_latency_sum")
	assert.NotContains(t, result, "# TYPE request_latency_count")
}

func TestFormatPrometheusText_HistogramAndGauge(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("request_latency_bucket", 10, map[string]string{"le": "0.1", "ip": "192.168.1.1", "port": "8080"}, now),
		createTestMetric("request_latency_sum", 12.5, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
		createTestMetric("request_latency_count", 30, map[string]string{"ip": "192.168.1.1", "port": "8080"}, now),
		createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "ip": "192.168.1.1", "port": "8080"}, now),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Contains(t, result, "# TYPE request_latency histogram")
	assert.Contains(t, result, "# TYPE cpu_usage gauge")
	assert.Contains(t, result, "request_latency_bucket{")
	assert.Contains(t, result, "cpu_usage{")
}

func TestGetHistogramBaseName(t *testing.T) {
	tests := []struct {
		name     string
		input    string
		expected string
	}{
		{"bucket suffix", "request_latency_bucket", "request_latency"},
		{"sum suffix", "request_latency_sum", "request_latency"},
		{"count suffix", "request_latency_count", "request_latency"},
		{"no histogram suffix", "cpu_usage", ""},
		{"empty string", "", ""},
		{"only suffix", "_bucket", ""},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := getHistogramBaseName(tt.input)
			assert.Equal(t, tt.expected, result)
		})
	}
}

func TestFormatMetricsWindowJSON_Empty(t *testing.T) {
	server := newTestServerOnly(t)

	result := server.formatMetricsWindowJSON([]*metrics.AggregatedMetric{})

	assert.Equal(t, 0, len(result))
}

func TestFormatMetricsWindowJSON_SingleMetric(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
	}

	result := server.formatMetricsWindowJSON(metricsList)

	assert.Equal(t, 1, len(result))
	item := result[0]
	assert.Equal(t, "cpu_usage", item["name"])
	assert.Equal(t, "test-agent-1", item["agent_id"])
	assert.Equal(t, "192.168.1.1", item["pod_name"])
	assert.Contains(t, item, "data")
}

func TestFormatMetricsWindowJSON_MultipleMetrics(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "pod_name": "192.168.1.1"}, now),
		createTestMetric("memory_usage", 50.0, map[string]string{"type": "heap", "pod_name": "192.168.1.1"}, now),
	}

	result := server.formatMetricsWindowJSON(metricsList)

	assert.Equal(t, 2, len(result))
	metricNames := make(map[string]bool)
	for _, item := range result {
		metricNames[item["name"].(string)] = true
	}
	assert.True(t, metricNames["cpu_usage"])
	assert.True(t, metricNames["memory_usage"])
}

func TestGetMetricKey(t *testing.T) {
	server := newTestServerOnly(t)

	metric := createTestMetric("cpu_usage", 75.5, map[string]string{"cpu": "0", "env": "test", "pod_name": "192.168.1.1"}, time.Now())

	key := server.getMetricKey(metric)

	assert.Contains(t, key, "cpu_usage")
	assert.Contains(t, key, "test-agent-1")
	assert.Contains(t, key, "cpu=0")
	assert.Contains(t, key, "env=test")
	assert.Contains(t, key, "pod_name=192.168.1.1")
}

func TestFormatFloat(t *testing.T) {
	tests := []struct {
		name     string
		expected string
		value    float64
	}{
		{"integer", "100", 100.0},
		{"decimal", "75.5", 75.5},
		{"small", "0.001", 0.001},
		{"large", "1000000", 1000000.0},
		{"negative", "-50.5", -50.5},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := formatFloat(tt.value)
			assert.Equal(t, tt.expected, result)
		})
	}
}

func TestServer_Integration(t *testing.T) {
	server := newTestServerOnly(t)

	mux := http.NewServeMux()
	mux.HandleFunc("/metrics", server.handleMetrics)
	mux.HandleFunc("/metrics-windows", server.handleMetricsWindows)
	mux.HandleFunc("/health", server.handleHealth)

	testServer := httptest.NewServer(mux)
	defer testServer.Close()

	client := &http.Client{Timeout: 500 * time.Millisecond}

	healthResp, healthErr := client.Get(testServer.URL + "/health")
	require.NoError(t, healthErr)
	assert.Equal(t, http.StatusOK, healthResp.StatusCode)
	healthResp.Body.Close()

	metricsResp, metricsErr := client.Get(testServer.URL + "/metrics")
	if metricsErr == nil {
		assert.Equal(t, http.StatusOK, metricsResp.StatusCode)
		metricsResp.Body.Close()
	}
}

func TestHandleMetricsWindows_WithFilters(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	startTime := now.Add(-1 * time.Hour).Format(time.RFC3339)
	endTime := now.Format(time.RFC3339)

	u, urlErr := url.Parse("/metrics-windows")
	require.NoError(t, urlErr)
	q := u.Query()
	q.Set("start_time", startTime)
	q.Set("end_time", endTime)
	q.Set("role", "datanode-cold")
	q.Set("address", "192.168.1.1")
	u.RawQuery = q.Encode()

	req := httptest.NewRequest(http.MethodGet, u.String(), nil)
	w := httptest.NewRecorder()

	server.handleMetricsWindows(w, req)

	assert.Equal(t, http.StatusOK, w.Code)
	var response []map[string]interface{}
	decodeErr := json.NewDecoder(w.Body).Decode(&response)
	require.NoError(t, decodeErr)
}

// mockLifecycleDataRequester is a mock implementation of lifecycle.RequestSender for testing.
type mockLifecycleDataRequester struct {
	requestErr   error
	lifecycleMgr *lifecycle.Manager
	dataByAgent  map[string]*fodcv1.LifecycleData
	podByAgent   map[string]string
}

func (m *mockLifecycleDataRequester) RequestLifecycleData(agentID string) error {
	if m.requestErr != nil {
		return m.requestErr
	}
	if data, exists := m.dataByAgent[agentID]; exists && m.lifecycleMgr != nil {
		podName := m.podByAgent[agentID]
		m.lifecycleMgr.UpdateLifecycle(agentID, podName, data)
	}
	return nil
}

func TestFormatPrometheusText_SortedMetrics(t *testing.T) {
	server := newTestServerOnly(t)

	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetric("z_metric", 1.0, map[string]string{"pod_name": "192.168.1.1"}, now),
		createTestMetric("a_metric", 2.0, map[string]string{"pod_name": "192.168.1.1"}, now),
		createTestMetric("m_metric", 3.0, map[string]string{"pod_name": "192.168.1.1"}, now),
	}

	result := server.formatPrometheusText(metricsList)

	aIndex := strings.Index(result, "a_metric")
	mIndex := strings.Index(result, "m_metric")
	zIndex := strings.Index(result, "z_metric")

	assert.True(t, aIndex < mIndex && mIndex < zIndex, "Metrics should be sorted alphabetically")
}

func TestHandleClusterLifecycle_Success(t *testing.T) {
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)

	mockLifecycleRequester := &mockLifecycleDataRequester{
		dataByAgent: make(map[string]*fodcv1.LifecycleData),
		podByAgent:  make(map[string]string),
	}
	lifecycleMgr := lifecycle.NewManager(testRegistry, mockLifecycleRequester, testLogger)
	mockLifecycleRequester.lifecycleMgr = lifecycleMgr

	ctx := context.Background()
	identity := registry.AgentIdentity{
		PodName:        "test-pod-1",
		Role:           "datanode",
		ContainerNames: []string{"banyandb"},
	}
	agentID, registerErr := testRegistry.RegisterAgent(ctx, identity)
	require.NoError(t, registerErr)

	mockLifecycleRequester.dataByAgent[agentID] = &fodcv1.LifecycleData{
		Reports: []*fodcv1.LifecycleReport{
			{Filename: "2026-03-26.json", ReportJson: `{"status":"ok"}`},
		},
		Groups: []*fodcv1.GroupLifecycleInfo{
			{Name: "group1", Catalog: "measure"},
		},
	}
	mockLifecycleRequester.podByAgent[agentID] = "test-pod-1"

	server := NewServer(aggregator, nil, lifecycleMgr, testRegistry, nil, nil, nil, testLogger)

	req := httptest.NewRequest(http.MethodGet, "/cluster/lifecycle", nil)
	resp := httptest.NewRecorder()
	server.handleClusterLifecycle(resp, req)

	assert.Equal(t, http.StatusOK, resp.Code)
	assert.Equal(t, "application/json", resp.Header().Get("Content-Type"))

	var result lifecycle.InspectionResult
	decodeErr := json.NewDecoder(resp.Body).Decode(&result)
	require.NoError(t, decodeErr)
	assert.Equal(t, 1, len(result.Groups))
	assert.Equal(t, "group1", result.Groups[0].Name)
	assert.Equal(t, 1, len(result.LifecycleStatuses))
	assert.Equal(t, "test-pod-1", result.LifecycleStatuses[0].PodName)
	assert.Equal(t, 1, len(result.LifecycleStatuses[0].Reports))
	assert.Equal(t, "2026-03-26.json", result.LifecycleStatuses[0].Reports[0].Filename)
}

func TestHandleClusterLifecycle_NoLifecycleManager(t *testing.T) {
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)
	server := NewServer(aggregator, nil, nil, testRegistry, nil, nil, nil, testLogger)

	req := httptest.NewRequest(http.MethodGet, "/cluster/lifecycle", nil)
	resp := httptest.NewRecorder()
	server.handleClusterLifecycle(resp, req)

	assert.Equal(t, http.StatusServiceUnavailable, resp.Code)
}

// TestHandleClusterLifecycle_NoAgents_FlagsErrorInBody reproduces the bug where the proxy
// silently returns 200 + an empty payload when no agents are registered. This makes the
// caller (e.g. the SRE inspection agent) unable to distinguish two semantically very
// different states:
//   - the BanyanDB cluster genuinely has no groups (a real cluster-side problem); or
//   - the FODC proxy lost all of its agents (an infrastructure-layer outage caused by a
//     proxy pod restart, every agent stuck on reconnect, etc.).
//
// In production the second case is what triggered the lifecycle critical alert, but the
// HTTP layer presented it identically to the first.
//
// Pre-fix behavior: handleClusterLifecycle writes 200 with body
// `{"groups":[],"lifecycle_statuses":[]}` and no way to tell the two cases apart.
// Post-fix behavior: HTTP status is still 200, but the body now carries an "error" field
// when agent_summary reports total=0 or responded=0, plus the agent_summary itself.
func TestHandleClusterLifecycle_NoAgents_FlagsErrorInBody(t *testing.T) {
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	defer testRegistry.Stop()
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)

	mockLifecycleRequester := &mockLifecycleDataRequester{
		dataByAgent: make(map[string]*fodcv1.LifecycleData),
		podByAgent:  make(map[string]string),
	}
	lifecycleMgr := lifecycle.NewManager(testRegistry, mockLifecycleRequester, testLogger)
	mockLifecycleRequester.lifecycleMgr = lifecycleMgr

	server := NewServer(aggregator, nil, lifecycleMgr, testRegistry, nil, nil, nil, testLogger)

	req := httptest.NewRequest(http.MethodGet, "/cluster/lifecycle", nil)
	resp := httptest.NewRecorder()
	server.handleClusterLifecycle(resp, req)

	assert.Equal(t, http.StatusOK, resp.Code, "status code stays 200 even when zero agents are registered")

	var body map[string]interface{}
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))

	errMsg, hasError := body["error"]
	assert.Truef(t, hasError, "body must include an error field when zero agents are registered, got: %v", body)
	assert.Containsf(t, errMsg, "FODC unavailable",
		"error field should describe the infrastructure-layer failure, got: %v", errMsg)

	summary, hasSummary := body["agent_summary"].(map[string]interface{})
	require.Truef(t, hasSummary, "body must include agent_summary, got: %v", body)
	assert.EqualValues(t, 0, summary["total"], "agent_summary.total should be 0 when no agents are registered")
	assert.EqualValues(t, 0, summary["responded"], "agent_summary.responded should be 0 when no agents are registered")
}

// runLifecycleHandlerWithGroup wires up a lifecycle manager backed by a single agent that
// reports the supplied group, invokes the HTTP handler, and returns the parsed response
// body so callers can inspect raw JSON keys.
func runLifecycleHandlerWithGroup(t *testing.T, group *fodcv1.GroupLifecycleInfo) map[string]interface{} {
	t.Helper()
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	t.Cleanup(testRegistry.Stop)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)

	mockLifecycleRequester := &mockLifecycleDataRequester{
		dataByAgent: make(map[string]*fodcv1.LifecycleData),
		podByAgent:  make(map[string]string),
	}
	lifecycleMgr := lifecycle.NewManager(testRegistry, mockLifecycleRequester, testLogger)
	mockLifecycleRequester.lifecycleMgr = lifecycleMgr

	identity := registry.AgentIdentity{
		PodName:        "test-pod-1",
		Role:           "datanode",
		ContainerNames: []string{"banyandb"},
	}
	agentID, registerErr := testRegistry.RegisterAgent(context.Background(), identity)
	require.NoError(t, registerErr)

	mockLifecycleRequester.dataByAgent[agentID] = &fodcv1.LifecycleData{
		Groups: []*fodcv1.GroupLifecycleInfo{group},
	}
	mockLifecycleRequester.podByAgent[agentID] = "test-pod-1"

	server := NewServer(aggregator, nil, lifecycleMgr, testRegistry, nil, nil, nil, testLogger)
	req := httptest.NewRequest(http.MethodGet, "/cluster/lifecycle", nil)
	resp := httptest.NewRecorder()
	server.handleClusterLifecycle(resp, req)
	require.Equal(t, http.StatusOK, resp.Code)

	var body map[string]interface{}
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
	return body
}

// firstGroup pulls the first group from the decoded response body and asserts the basic shape.
func firstGroup(t *testing.T, body map[string]interface{}) map[string]interface{} {
	t.Helper()
	groups, ok := body["groups"].([]interface{})
	require.True(t, ok, "response body must contain a groups array")
	require.NotEmpty(t, groups, "groups array must not be empty")
	g, ok := groups[0].(map[string]interface{})
	require.True(t, ok, "first group must decode to an object")
	return g
}

func TestHandleClusterLifecycle_UsesSnakeCase(t *testing.T) {
	group := &fodcv1.GroupLifecycleInfo{
		Name: "g1",
		ResourceOpts: &commonv1.ResourceOpts{
			ShardNum: 4,
		},
	}
	body := runLifecycleHandlerWithGroup(t, group)
	g := firstGroup(t, body)

	resourceOpts, ok := g["resource_opts"].(map[string]interface{})
	require.True(t, ok, "field name must be snake_case 'resource_opts', not 'resourceOpts'")
	assert.Equal(t, float64(4), resourceOpts["shard_num"], "field name must be snake_case 'shard_num'")
	_, hasCamel := g["resourceOpts"]
	assert.False(t, hasCamel, "camelCase key must not be emitted")
}

func TestHandleClusterLifecycle_EmitsZeroValueFields(t *testing.T) {
	group := &fodcv1.GroupLifecycleInfo{
		Name: "g1",
		ResourceOpts: &commonv1.ResourceOpts{
			ShardNum: 4,
			Replicas: 0, // zero value must still appear in JSON output
			Stages: []*commonv1.LifecycleStage{
				{
					Name:     "warm",
					ShardNum: 2,
					Close:    false, // zero value must still appear
					Replicas: 0,     // zero value must still appear
				},
			},
		},
	}
	body := runLifecycleHandlerWithGroup(t, group)
	g := firstGroup(t, body)

	resourceOpts := g["resource_opts"].(map[string]interface{})
	_, hasReplicas := resourceOpts["replicas"]
	assert.True(t, hasReplicas, "zero-value replicas at ResourceOpts level must be emitted, not omitted")
	assert.Equal(t, float64(0), resourceOpts["replicas"])

	stages, ok := resourceOpts["stages"].([]interface{})
	require.True(t, ok)
	require.Len(t, stages, 1)
	stage := stages[0].(map[string]interface{})

	_, hasClose := stage["close"]
	assert.True(t, hasClose, "zero-value close=false in LifecycleStage must be emitted")
	assert.Equal(t, false, stage["close"])

	_, hasStageReplicas := stage["replicas"]
	assert.True(t, hasStageReplicas, "zero-value replicas in LifecycleStage must be emitted")
	assert.Equal(t, float64(0), stage["replicas"])
}

func TestHandleClusterLifecycle_NilResourceOptsIsNull(t *testing.T) {
	group := &fodcv1.GroupLifecycleInfo{
		Name: "g1",
		// ResourceOpts intentionally nil
	}
	body := runLifecycleHandlerWithGroup(t, group)
	g := firstGroup(t, body)

	require.Contains(t, g, "resource_opts", "EmitUnpopulated must surface the absent message as a key")
	assert.Nil(t, g["resource_opts"], "nil protobuf message marshals to JSON null under EmitUnpopulated")
}

func TestHandleClusterLifecycle_DataInfoEmitted(t *testing.T) {
	group := &fodcv1.GroupLifecycleInfo{
		Name: "g1",
		DataInfo: []*databasev1.DataInfo{
			{DataSizeBytes: 1024},
		},
	}
	body := runLifecycleHandlerWithGroup(t, group)
	g := firstGroup(t, body)

	dataInfo, ok := g["data_info"].([]interface{})
	require.True(t, ok, "data_info must be encoded as snake_case array")
	require.Len(t, dataInfo, 1)
}

// TestFormatPrometheusText_CounterType verifies that a metric with type=counter emits
// "# TYPE foo_total counter" and is NOT mistaken for a histogram (invariant 2).
func TestFormatPrometheusText_CounterType(t *testing.T) {
	server := newTestServerOnly(t)
	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetricWithType("http_requests_total", 42, map[string]string{"method": "GET"}, now, "counter"),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Contains(t, result, "# TYPE http_requests_total counter")
	assert.Contains(t, result, "http_requests_total{")
	assert.NotContains(t, result, "# TYPE http_requests_total gauge")
	assert.NotContains(t, result, "# TYPE http_requests_total histogram")
}

// TestFormatPrometheusText_CounterWithCountSuffix verifies that a counter named foo_count
// (not a histogram component) keeps type=counter and is NOT folded into histogram base "foo"
// (invariant 3 — the key fix for the mislabeled-histogram defect).
func TestFormatPrometheusText_CounterWithCountSuffix(t *testing.T) {
	server := newTestServerOnly(t)
	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetricWithType("banyandb_grpc_request_count", 100, map[string]string{"pod_name": "p1"}, now, "counter"),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Contains(t, result, "# TYPE banyandb_grpc_request_count counter")
	// Must NOT be folded into a histogram base called "banyandb_grpc_request"
	assert.NotContains(t, result, "# TYPE banyandb_grpc_request histogram")
	assert.Contains(t, result, "banyandb_grpc_request_count{")
}

// TestFormatPrometheusText_HistogramTyped verifies that histogram component series with
// type=histogram share exactly one "# TYPE foo histogram" line (invariant 4).
func TestFormatPrometheusText_HistogramTyped(t *testing.T) {
	server := newTestServerOnly(t)
	now := time.Now()
	labels := map[string]string{"pod_name": "p1"}
	metricsList := []*metrics.AggregatedMetric{
		createTestMetricWithType("request_latency_bucket", 10, labels, now, "histogram"),
		createTestMetricWithType("request_latency_sum", 1.5, labels, now, "histogram"),
		createTestMetricWithType("request_latency_count", 10, labels, now, "histogram"),
	}

	result := server.formatPrometheusText(metricsList)

	// Exactly one TYPE line for the base name.
	assert.Equal(t, 1, strings.Count(result, "# TYPE request_latency histogram"),
		"must emit exactly one TYPE histogram line")
	assert.Contains(t, result, "request_latency_bucket{")
	assert.Contains(t, result, "request_latency_sum{")
	assert.Contains(t, result, "request_latency_count{")
	// Component names must not get their own TYPE lines.
	assert.NotContains(t, result, "# TYPE request_latency_bucket")
	assert.NotContains(t, result, "# TYPE request_latency_sum")
	assert.NotContains(t, result, "# TYPE request_latency_count")
}

// TestFormatPrometheusText_SummaryTyped verifies that summary families emit exactly one
// "# TYPE foo summary" line and all series (quantile, _sum, _count) fall under it (invariant 5).
func TestFormatPrometheusText_SummaryTyped(t *testing.T) {
	server := newTestServerOnly(t)
	now := time.Now()
	labelsQ := map[string]string{"quantile": "0.5"}
	labelsBase := map[string]string{}
	metricsList := []*metrics.AggregatedMetric{
		createTestMetricWithType("go_gc_duration_seconds", 0.0001, labelsQ, now, "summary"),
		createTestMetricWithType("go_gc_duration_seconds_sum", 0.5, labelsBase, now, "summary"),
		createTestMetricWithType("go_gc_duration_seconds_count", 42, labelsBase, now, "summary"),
	}

	result := server.formatPrometheusText(metricsList)

	// Exactly one TYPE line for the base name.
	assert.Equal(t, 1, strings.Count(result, "# TYPE go_gc_duration_seconds summary"),
		"must emit exactly one TYPE summary line")
	assert.Contains(t, result, "go_gc_duration_seconds{")
	assert.Contains(t, result, "go_gc_duration_seconds_sum")
	assert.Contains(t, result, "go_gc_duration_seconds_count")
	// Must NOT emit a separate gauge TYPE line for the bare name.
	assert.NotContains(t, result, "# TYPE go_gc_duration_seconds gauge")
}

// TestFormatPrometheusText_MixedVersionDedup verifies that when the same metric family
// arrives both typed (upgraded agent) and untyped (pre-upgrade agent) in one scrape, the
// proxy emits exactly ONE "# TYPE" line — the typed one — and folds the untyped samples
// under it, rather than emitting a conflicting second TYPE line that Prometheus rejects.
func TestFormatPrometheusText_MixedVersionDedup(t *testing.T) {
	server := newTestServerOnly(t)
	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetricWithType("process_cpu_seconds_total", 5, map[string]string{"pod_name": "new"}, now, "counter"),
		createTestMetricWithType("process_cpu_seconds_total", 3, map[string]string{"pod_name": "old"}, now, ""),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Equal(t, 1, strings.Count(result, "# TYPE process_cpu_seconds_total"),
		"must emit exactly one TYPE line for the family")
	assert.Contains(t, result, "# TYPE process_cpu_seconds_total counter")
	assert.NotContains(t, result, "# TYPE process_cpu_seconds_total gauge")
	// Both the upgraded and pre-upgrade samples must still be present.
	assert.Contains(t, result, `pod_name="new"`)
	assert.Contains(t, result, `pod_name="old"`)
}

// TestFormatPrometheusText_MixedVersionHistogramDedup verifies the same dedup for histogram
// component series: untyped _bucket/_sum/_count samples from a pre-upgrade agent fold under
// the typed histogram family rather than spawning a second (conflicting) TYPE line.
func TestFormatPrometheusText_MixedVersionHistogramDedup(t *testing.T) {
	server := newTestServerOnly(t)
	now := time.Now()
	newPod := map[string]string{"le": "0.1", "pod_name": "new"}
	oldPod := map[string]string{"le": "0.1", "pod_name": "old"}
	metricsList := []*metrics.AggregatedMetric{
		createTestMetricWithType("req_latency_bucket", 10, newPod, now, "histogram"),
		createTestMetricWithType("req_latency_sum", 1.0, map[string]string{"pod_name": "new"}, now, "histogram"),
		createTestMetricWithType("req_latency_count", 10, map[string]string{"pod_name": "new"}, now, "histogram"),
		createTestMetricWithType("req_latency_bucket", 8, oldPod, now, ""),
		createTestMetricWithType("req_latency_sum", 0.8, map[string]string{"pod_name": "old"}, now, ""),
		createTestMetricWithType("req_latency_count", 8, map[string]string{"pod_name": "old"}, now, ""),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Equal(t, 1, strings.Count(result, "# TYPE req_latency histogram"),
		"must emit exactly one histogram TYPE line for the family")
	// No component or gauge TYPE lines leaked from the legacy path.
	assert.NotContains(t, result, "# TYPE req_latency_bucket")
	assert.NotContains(t, result, "# TYPE req_latency_sum")
	assert.NotContains(t, result, "# TYPE req_latency_count")
	assert.NotContains(t, result, "# TYPE req_latency gauge")
	// Pre-upgrade samples still present.
	assert.Contains(t, result, `pod_name="old"`)
}

// TestFormatPrometheusText_EscapesLabelValues verifies that label values containing the
// Prometheus-special characters (backslash, double-quote, line feed) are escaped, so a value
// with special characters cannot produce malformed exposition output that breaks the scrape.
func TestFormatPrometheusText_EscapesLabelValues(t *testing.T) {
	server := newTestServerOnly(t)
	now := time.Now()
	metricsList := []*metrics.AggregatedMetric{
		createTestMetricWithType("http_requests_total", 1, map[string]string{
			"path": `a"b\c` + "\n" + "d",
		}, now, "counter"),
	}

	result := server.formatPrometheusText(metricsList)

	assert.Contains(t, result, `path="a\"b\\c\nd"`)
	// The raw, unescaped sequence must not leak into the output.
	assert.NotContains(t, result, `path="a"b\c`)
}

func TestHandleClusterLifecycle_StatusReportsUseProtoJSON(t *testing.T) {
	initTestLogger(t)
	testLogger := logger.GetLogger("test", "api")
	testRegistry := registry.NewAgentRegistry(testLogger, 5*time.Second, 10*time.Second, 100)
	t.Cleanup(testRegistry.Stop)
	mockSender := &mockRequestSender{}
	aggregator := metrics.NewAggregator(testRegistry, mockSender, testLogger)

	mockLifecycleRequester := &mockLifecycleDataRequester{
		dataByAgent: make(map[string]*fodcv1.LifecycleData),
		podByAgent:  make(map[string]string),
	}
	lifecycleMgr := lifecycle.NewManager(testRegistry, mockLifecycleRequester, testLogger)
	mockLifecycleRequester.lifecycleMgr = lifecycleMgr

	identity := registry.AgentIdentity{
		PodName:        "test-pod-1",
		Role:           "datanode",
		ContainerNames: []string{"banyandb"},
	}
	agentID, registerErr := testRegistry.RegisterAgent(context.Background(), identity)
	require.NoError(t, registerErr)

	mockLifecycleRequester.dataByAgent[agentID] = &fodcv1.LifecycleData{
		Reports: []*fodcv1.LifecycleReport{
			{Filename: "2026-04-29.json", ReportJson: `{"status":"ok"}`},
			{Filename: "2026-04-28.json", ReportJson: ""}, // empty value must still be emitted
		},
	}
	mockLifecycleRequester.podByAgent[agentID] = "test-pod-1"

	server := NewServer(aggregator, nil, lifecycleMgr, testRegistry, nil, nil, nil, testLogger)
	req := httptest.NewRequest(http.MethodGet, "/cluster/lifecycle", nil)
	resp := httptest.NewRecorder()
	server.handleClusterLifecycle(resp, req)
	require.Equal(t, http.StatusOK, resp.Code)

	var body map[string]interface{}
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))

	statuses, ok := body["lifecycle_statuses"].([]interface{})
	require.True(t, ok)
	require.Len(t, statuses, 1)
	status := statuses[0].(map[string]interface{})
	assert.Equal(t, "test-pod-1", status["pod_name"])

	reports, ok := status["reports"].([]interface{})
	require.True(t, ok)
	require.Len(t, reports, 2)

	first := reports[0].(map[string]interface{})
	assert.Equal(t, "2026-04-29.json", first["filename"])
	assert.Equal(t, `{"status":"ok"}`, first["report_json"], "field must be snake_case 'report_json'")

	second := reports[1].(map[string]interface{})
	_, hasReportJSON := second["report_json"]
	assert.True(t, hasReportJSON, "empty report_json must still be emitted under EmitUnpopulated")
	assert.Equal(t, "", second["report_json"])
}
