package pkg

import (
	"bytes"
	"compress/gzip"
	"context"
	"crypto/rand"
	"fmt"
	"io"
	"mime"
	"mime/multipart"
	"net"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/andybalholm/brotli"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.keploy.io/server/v3/internal/testnet"
	"go.keploy.io/server/v3/pkg/models"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

func TestNextIDIgnoresOtherPrefixes(t *testing.T) {
	tests := []struct {
		name       string
		ids        []string
		identifier string
		want       string
	}{
		{
			name:       "custom metadata name with numeric suffix does not affect test-set sequence",
			ids:        []string{"test-set-0", "checkout-flow-999"},
			identifier: models.TestSetPattern,
			want:       "test-set-1",
		},
		{
			name:       "test-set IDs do not affect test-run sequence",
			ids:        []string{"test-run-2", "test-set-99"},
			identifier: models.TestRunTemplateName,
			want:       "test-run-3",
		},
		{
			name:       "invalid suffix is ignored",
			ids:        []string{"test-set-0", "test-set-0 copy", "test-set-final"},
			identifier: models.TestSetPattern,
			want:       "test-set-1",
		},
		{
			name:       "signed suffixes are not indices",
			ids:        []string{"test-set-0", "test-set--1", "test-set-+5"},
			identifier: models.TestSetPattern,
			want:       "test-set-1",
		},
		{
			name:       "empty set starts at zero",
			ids:        nil,
			identifier: models.TestSetPattern,
			want:       "test-set-0",
		},
		{
			name:       "bare prefix is not an index",
			ids:        []string{"test-set-"},
			identifier: models.TestSetPattern,
			want:       "test-set-0",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			assert.Equal(t, tt.want, NextID(tt.ids, tt.identifier))
		})
	}
}

func TestLastIDIgnoresOtherPrefixes(t *testing.T) {
	tests := []struct {
		name       string
		ids        []string
		identifier string
		want       string
	}{
		{
			name:       "custom metadata name with numeric suffix does not affect latest test run",
			ids:        []string{"test-run-2", "checkout-flow-999"},
			identifier: models.TestRunTemplateName,
			want:       "test-run-2",
		},
		{
			name:       "test runs do not affect latest test set",
			ids:        []string{"test-set-1", "test-run-99"},
			identifier: models.TestSetPattern,
			want:       "test-set-1",
		},
		{
			name:       "invalid suffix is ignored",
			ids:        []string{"test-run-3", "test-run-3 copy", "test-run-final"},
			identifier: models.TestRunTemplateName,
			want:       "test-run-3",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			assert.Equal(t, tt.want, LastID(tt.ids, tt.identifier))
		})
	}
}

// TestSimulateHTTP_NewRequestError_303 ensures that SimulateHTTP returns an error
// when http.NewRequestWithContext fails. This is triggered by providing an invalid
// HTTP method string in the test case.
func TestSimulateHTTP_NewRequestError_303(t *testing.T) {
	// Arrange
	ctx := context.Background()
	logger := zap.NewNop()
	tc := &models.TestCase{
		Name: "test-case-new-request-error",
		HTTPReq: models.HTTPReq{
			Method: "INVALID METHOD", // Invalid method
			URL:    "http://example.com/test",
			Body:   `{"key":"value"}`,
		},
	}

	// Act
	resp, err := SimulateHTTP(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 10})

	// Assert
	require.Error(t, err)
	assert.Contains(t, err.Error(), "invalid method")
	assert.Nil(t, resp)
}

func TestSimulateHTTP_MultipartRebuildWithPaths_314(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	// The file is kept in the recording, and named from its directory: a
	// test case's file is read only there (ReadRecordingFile).
	recording := t.TempDir()
	tempFile, err := os.CreateTemp(recording, "keploy-multipart-test-*.txt")
	require.NoError(t, err)
	_, err = tempFile.WriteString("file-data")
	require.NoError(t, err)
	require.NoError(t, tempFile.Close())

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		contentType := r.Header.Get("Content-Type")
		mediaType, params, err := mime.ParseMediaType(contentType)
		if err != nil {
			t.Errorf("failed to parse media type: %v", err)
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		assert.Equal(t, "multipart/form-data", mediaType)
		if params["boundary"] == "" {
			t.Errorf("missing multipart boundary")
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		assert.NotEqual(t, "test-boundary", params["boundary"])

		if cl := r.Header.Get("Content-Length"); cl != "" {
			assert.NotEqual(t, "1", cl)
		}

		reader := multipart.NewReader(r.Body, params["boundary"])
		fields := map[string]string{}
		files := map[string]string{}
		fileNames := map[string]string{}
		for {
			part, err := reader.NextPart()
			if err == io.EOF {
				break
			}
			if err != nil {
				t.Errorf("failed to read multipart part: %v", err)
				w.WriteHeader(http.StatusBadRequest)
				return
			}
			name := part.FormName()
			data, err := io.ReadAll(part)
			if err != nil {
				t.Errorf("failed to read multipart part data: %v", err)
				w.WriteHeader(http.StatusBadRequest)
				return
			}
			if part.FileName() != "" {
				files[name] = string(data)
				fileNames[name] = part.FileName()
				continue
			}
			fields[name] = string(data)
		}

		assert.Equal(t, "text-value", fields["text"])
		assert.Equal(t, "file-data", files["upload"])
		assert.Equal(t, filepath.Base(tempFile.Name()), fileNames["upload"])

		w.WriteHeader(http.StatusOK)
	}))
	defer server.Close()

	tc := &models.TestCase{
		Name: "multipart-paths",
		HTTPReq: models.HTTPReq{
			Method: "POST",
			URL:    server.URL,
			Header: map[string]string{
				"Content-Type":   "multipart/form-data; boundary=test-boundary",
				"Content-Length": "1",
			},
			Form: []models.FormData{
				{
					Key:    "text",
					Values: []string{"text-value"},
				},
				{
					Key:   "upload",
					Paths: []string{filepath.Base(tempFile.Name())},
				},
			},
		},
	}

	resp, err := SimulateHTTP(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 10, KeployPath: recording})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, http.StatusOK, resp.StatusCode)
}

func TestSimulateHTTP_MultipartRebuildWithFileNames_315(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		contentType := r.Header.Get("Content-Type")
		mediaType, params, err := mime.ParseMediaType(contentType)
		if err != nil {
			t.Errorf("failed to parse media type: %v", err)
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		assert.Equal(t, "multipart/form-data", mediaType)
		if params["boundary"] == "" {
			t.Errorf("missing multipart boundary")
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		assert.NotEqual(t, "legacy-boundary", params["boundary"])

		reader := multipart.NewReader(r.Body, params["boundary"])
		fields := map[string]string{}
		files := map[string]string{}
		fileNames := map[string]string{}
		for {
			part, err := reader.NextPart()
			if err == io.EOF {
				break
			}
			if err != nil {
				t.Errorf("failed to read multipart part: %v", err)
				w.WriteHeader(http.StatusBadRequest)
				return
			}
			name := part.FormName()
			data, err := io.ReadAll(part)
			if err != nil {
				t.Errorf("failed to read multipart part data: %v", err)
				w.WriteHeader(http.StatusBadRequest)
				return
			}
			if part.FileName() != "" {
				files[name] = string(data)
				fileNames[name] = part.FileName()
				continue
			}
			fields[name] = string(data)
		}

		assert.Equal(t, "text-value", fields["text"])
		assert.Equal(t, "binary-content", files["payload"])
		assert.Equal(t, "blob.bin", fileNames["payload"])

		w.WriteHeader(http.StatusOK)
	}))
	defer server.Close()

	tc := &models.TestCase{
		Name: "multipart-filenames",
		HTTPReq: models.HTTPReq{
			Method: "POST",
			URL:    server.URL,
			Header: map[string]string{
				"Content-Type": "multipart/form-data; boundary=legacy-boundary",
			},
			Form: []models.FormData{
				{
					Key:    "text",
					Values: []string{"text-value"},
				},
				{
					Key:       "payload",
					Values:    []string{"binary-content"},
					FileNames: []string{"blob.bin"},
				},
			},
		},
	}

	resp, err := SimulateHTTP(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 10})
	require.NoError(t, err)
	require.NotNil(t, resp)
	assert.Equal(t, http.StatusOK, resp.StatusCode)
}

func TestSimulateHTTP_SSEStreamMatchAndEarlyClose_316(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	serverClosed := make(chan struct{})
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
		flusher, ok := w.(http.Flusher)
		require.True(t, ok, "server response writer must support flushing")

		_, _ = w.Write([]byte("id:100\nevent:ticker\ndata:{\"value\":1}\n\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("id:101\nevent:ticker\ndata:{\"value\":2}\n\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("id:102\nevent:ticker\ndata:{\"value\":3}\n\n"))
		flusher.Flush()

		<-r.Context().Done()
		close(serverClosed)
	}))
	defer server.Close()

	expectedSSEBody := strings.Join([]string{
		"id:100",
		"event:ticker",
		"data:{\"value\":1}",
		"",
		"id:101",
		"event:ticker",
		"data:{\"value\":2}",
		"",
	}, "\n")

	tc := &models.TestCase{
		Name: "sse-match-and-close",
		HTTPReq: models.HTTPReq{
			Method: "GET",
			URL:    server.URL,
			Header: map[string]string{
				"Accept": "text/event-stream",
			},
		},
		HTTPResp: models.HTTPResp{
			Header: map[string]string{
				"Content-Type": "text/event-stream; charset=utf-8",
			},
			Body: expectedSSEBody,
			StreamBody: []models.HTTPStreamChunk{
				{
					Data: []models.HTTPStreamDataField{
						{Key: "id", Value: "100"},
						{Key: "event", Value: "ticker"},
						{Key: "data", Value: `{"value":1}`},
					},
				},
				{
					Data: []models.HTTPStreamDataField{
						{Key: "id", Value: "101"},
						{Key: "event", Value: "ticker"},
						{Key: "data", Value: `{"value":2}`},
					},
				},
			},
		},
	}

	// Use SimulateHTTPStreaming for streaming test cases
	streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 3})
	require.NoError(t, err)
	require.NotNil(t, streamResp)

	// Compare the stream
	noiseKeys := map[string]struct{}{}
	matched, capturedBody, _, compareErr := CompareHTTPStream(tc.HTTPResp, streamResp.Reader, streamResp.StreamConfig, noiseKeys, logger)
	streamResp.Reader.Close()
	require.NoError(t, compareErr)
	require.True(t, matched)

	respBody := tc.HTTPResp.Body
	if !matched {
		respBody = capturedBody
	}

	assert.Equal(t, http.StatusOK, streamResp.StatusCode)
	assert.Equal(t, expectedSSEBody, respBody)

	select {
	case <-serverClosed:
	case <-time.After(1 * time.Second):
		t.Fatal("expected server stream to be closed by client after matching SSE queue")
	}
}

func TestSimulateHTTP_SSEStreamMismatch_317(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		flusher, ok := w.(http.Flusher)
		require.True(t, ok, "server response writer must support flushing")

		_, _ = w.Write([]byte("id:1\nevent:update\ndata:{\"value\":1}\n\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("id:2\nevent:update\ndata:{\"value\":999}\n\n"))
		flusher.Flush()
	}))
	defer server.Close()

	tc := &models.TestCase{
		Name: "sse-mismatch",
		HTTPReq: models.HTTPReq{
			Method: "GET",
			URL:    server.URL,
			Header: map[string]string{
				"Accept": "text/event-stream",
			},
		},
		HTTPResp: models.HTTPResp{
			Header: map[string]string{
				"Content-Type": "text/event-stream",
			},
			Body: strings.Join([]string{
				"id:1",
				"event:update",
				"data:{\"value\":1}",
				"",
				"id:2",
				"event:update",
				"data:{\"value\":2}",
				"",
			}, "\n"),
			StreamBody: []models.HTTPStreamChunk{
				{
					Data: []models.HTTPStreamDataField{
						{Key: "id", Value: "1"},
						{Key: "event", Value: "update"},
						{Key: "data", Value: `{"value":1}`},
					},
				},
				{
					Data: []models.HTTPStreamDataField{
						{Key: "id", Value: "2"},
						{Key: "event", Value: "update"},
						{Key: "data", Value: `{"value":2}`},
					},
				},
			},
		},
	}

	// Use SimulateHTTPStreaming for streaming test cases
	streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 3})
	require.NoError(t, err)
	require.NotNil(t, streamResp)

	// Compare the stream - should NOT match due to value mismatch
	noiseKeys := map[string]struct{}{}
	matched, capturedBody, _, compareErr := CompareHTTPStream(tc.HTTPResp, streamResp.Reader, streamResp.StreamConfig, noiseKeys, logger)
	streamResp.Reader.Close()
	require.NoError(t, compareErr)

	respBody := capturedBody
	if matched {
		respBody = tc.HTTPResp.Body
	}

	assert.Equal(t, http.StatusOK, streamResp.StatusCode)
	assert.NotEqual(t, tc.HTTPResp.Body, respBody)
	assert.Contains(t, respBody, "999")
}

func TestSimulateHTTP_SSEStreamMatch_WithStructuredExpectedBody_317A(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		flusher, ok := w.(http.Flusher)
		require.True(t, ok, "server response writer must support flushing")

		_, _ = w.Write([]byte("id:1\nevent:update\ndata:{\"value\":1}\n\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("id:2\nevent:update\ndata:{\"value\":2}\n\n"))
		flusher.Flush()
	}))
	defer server.Close()

	tc := &models.TestCase{
		Name: "sse-structured-expected",
		HTTPReq: models.HTTPReq{
			Method: "GET",
			URL:    server.URL,
			Header: map[string]string{
				"Accept": "text/event-stream",
			},
		},
		HTTPResp: models.HTTPResp{
			Header: map[string]string{
				"Content-Type": "text/event-stream",
			},
			// This intentionally does not match the streamed frames. The stream
			// comparator must use HTTPResp.StreamBody when present.
			Body: "legacy-body-not-used-for-stream-compare",
			StreamBody: []models.HTTPStreamChunk{
				{
					Data: []models.HTTPStreamDataField{
						{Key: "id", Value: "1"},
						{Key: "event", Value: "update"},
						{Key: "data", Value: `{"value":1}`},
					},
				},
				{
					Data: []models.HTTPStreamDataField{
						{Key: "id", Value: "2"},
						{Key: "event", Value: "update"},
						{Key: "data", Value: `{"value":2}`},
					},
				},
			},
		},
	}

	// Use SimulateHTTPStreaming for streaming test cases
	streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 3})
	require.NoError(t, err)
	require.NotNil(t, streamResp)

	// Compare the stream
	noiseKeys := map[string]struct{}{}
	matched, _, _, compareErr := CompareHTTPStream(tc.HTTPResp, streamResp.Reader, streamResp.StreamConfig, noiseKeys, logger)
	streamResp.Reader.Close()
	require.NoError(t, compareErr)
	require.True(t, matched)

	// When matched, use tc.HTTPResp.Body (legacy body)
	respBody := tc.HTTPResp.Body

	assert.Equal(t, http.StatusOK, streamResp.StatusCode)
	assert.Equal(t, tc.HTTPResp.Body, respBody)
}

func TestCanonicalizeSSEFrame_318(t *testing.T) {
	input := "id: 100\nevent: system-alert\ndata: {\"active\": true, \"user\" : \"alice\"}\n"
	got := canonicalizeSSEFrame(input)

	assert.Equal(t, "id:100\nevent:system-alert\ndata:{\"active\":true,\"user\":\"alice\"}", got)
}

func TestSimulateHTTP_NDJSONStreamMatch_WithStructuredExpectedBody_318A(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "application/x-ndjson")
		flusher, ok := w.(http.Flusher)
		require.True(t, ok, "server response writer must support flushing")

		_, _ = w.Write([]byte("{\"id\":1,\"ok\":true}\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("{\"id\":2,\"ok\":false}\n"))
		flusher.Flush()
	}))
	defer server.Close()

	tc := &models.TestCase{
		Name: "ndjson-structured-expected",
		HTTPReq: models.HTTPReq{
			Method: "GET",
			URL:    server.URL,
		},
		HTTPResp: models.HTTPResp{
			Header: map[string]string{
				"Content-Type": "application/x-ndjson",
			},
			Body: "legacy-body-not-used-for-stream-compare",
			StreamBody: []models.HTTPStreamChunk{
				{
					Data: []models.HTTPStreamDataField{
						{Key: "raw", Value: `{"id":1,"ok":true}`},
					},
				},
				{
					Data: []models.HTTPStreamDataField{
						{Key: "raw", Value: `{"id":2,"ok":false}`},
					},
				},
			},
		},
	}

	// Use SimulateHTTPStreaming for streaming test cases
	streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 3})
	require.NoError(t, err)
	require.NotNil(t, streamResp)

	// Compare the stream
	noiseKeys := map[string]struct{}{}
	matched, _, _, compareErr := CompareHTTPStream(tc.HTTPResp, streamResp.Reader, streamResp.StreamConfig, noiseKeys, logger)
	streamResp.Reader.Close()
	require.NoError(t, compareErr)
	require.True(t, matched)

	// When matched, use tc.HTTPResp.Body (legacy body)
	respBody := tc.HTTPResp.Body

	assert.Equal(t, http.StatusOK, streamResp.StatusCode)
	assert.Equal(t, tc.HTTPResp.Body, respBody)
}

func TestSimulateHTTP_NDJSONStreamMatchAndEarlyClose_319(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	serverClosed := make(chan struct{})
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/x-ndjson")
		flusher, ok := w.(http.Flusher)
		require.True(t, ok, "server response writer must support flushing")

		_, _ = w.Write([]byte("{\"id\":1,\"ok\":true}\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("{\"id\":2,\"ok\":false}\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("{\"id\":3,\"ok\":true}\n"))
		flusher.Flush()

		<-r.Context().Done()
		close(serverClosed)
	}))
	defer server.Close()

	expectedBody := "{\"id\":1,\"ok\":true}\n{\"id\":2,\"ok\":false}\n"

	tc := &models.TestCase{
		Name: "ndjson-match-and-close",
		HTTPReq: models.HTTPReq{
			Method: "GET",
			URL:    server.URL,
		},
		HTTPResp: models.HTTPResp{
			Header: map[string]string{
				"Content-Type": "application/x-ndjson",
			},
			Body: expectedBody,
			StreamBody: []models.HTTPStreamChunk{
				{
					Data: []models.HTTPStreamDataField{
						{Key: "raw", Value: `{"id":1,"ok":true}`},
					},
				},
				{
					Data: []models.HTTPStreamDataField{
						{Key: "raw", Value: `{"id":2,"ok":false}`},
					},
				},
			},
		},
	}

	// Use SimulateHTTPStreaming for streaming test cases
	streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 3})
	require.NoError(t, err)
	require.NotNil(t, streamResp)

	// Compare the stream
	noiseKeys := map[string]struct{}{}
	matched, _, _, compareErr := CompareHTTPStream(tc.HTTPResp, streamResp.Reader, streamResp.StreamConfig, noiseKeys, logger)
	streamResp.Reader.Close()
	require.NoError(t, compareErr)
	require.True(t, matched)

	respBody := expectedBody
	if !matched {
		respBody = ""
	}

	assert.Equal(t, http.StatusOK, streamResp.StatusCode)
	assert.Equal(t, expectedBody, respBody)

	select {
	case <-serverClosed:
	case <-time.After(1 * time.Second):
		t.Fatal("expected NDJSON stream to be closed by client after matching queue")
	}
}

func TestSimulateHTTP_MultipartStreamMatchAndEarlyClose_320(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()
	const boundary = "myCustomBoundary"

	writePart := func(w http.ResponseWriter, contentType string, body string) {
		_, _ = w.Write([]byte("--" + boundary + "\r\n"))
		_, _ = w.Write([]byte("Content-Type: " + contentType + "\r\n\r\n"))
		_, _ = w.Write([]byte(body))
		_, _ = w.Write([]byte("\r\n"))
	}

	serverClosed := make(chan struct{})
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary="+boundary)
		flusher, ok := w.(http.Flusher)
		require.True(t, ok, "server response writer must support flushing")

		writePart(w, "application/json", `{"frame":1}`)
		flusher.Flush()
		writePart(w, "text/plain", "frame-2")
		flusher.Flush()
		writePart(w, "text/plain", "frame-3")
		flusher.Flush()

		<-r.Context().Done()
		close(serverClosed)
	}))
	defer server.Close()

	expectedBody := strings.Join([]string{
		"--" + boundary,
		"Content-Type: application/json",
		"",
		`{"frame":1}`,
		"--" + boundary,
		"Content-Type: text/plain",
		"",
		"frame-2",
		"",
	}, "\r\n")

	tc := &models.TestCase{
		Name: "multipart-match-and-close",
		HTTPReq: models.HTTPReq{
			Method: "GET",
			URL:    server.URL,
		},
		HTTPResp: models.HTTPResp{
			Header: map[string]string{
				"Content-Type": "multipart/x-mixed-replace; boundary=" + boundary,
			},
			Body: expectedBody,
			StreamBody: []models.HTTPStreamChunk{
				{
					Data: []models.HTTPStreamDataField{
						{Key: "raw", Value: expectedBody},
					},
				},
			},
		},
	}

	// Use SimulateHTTPStreaming for streaming test cases
	streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 3})
	require.NoError(t, err)
	require.NotNil(t, streamResp)

	// Compare the stream
	noiseKeys := map[string]struct{}{}
	matched, _, _, compareErr := CompareHTTPStream(tc.HTTPResp, streamResp.Reader, streamResp.StreamConfig, noiseKeys, logger)
	streamResp.Reader.Close()
	require.NoError(t, compareErr)
	require.True(t, matched)

	respBody := expectedBody
	if !matched {
		respBody = ""
	}

	assert.Equal(t, http.StatusOK, streamResp.StatusCode)
	assert.Equal(t, expectedBody, respBody)

	select {
	case <-serverClosed:
	case <-time.After(1 * time.Second):
		t.Fatal("expected multipart stream to be closed by client after matching queue")
	}
}

func TestSimulateHTTP_PlainTextStreamMatchAndEarlyClose_321(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()

	serverClosed := make(chan struct{})
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/plain")
		flusher, ok := w.(http.Flusher)
		require.True(t, ok, "server response writer must support flushing")

		_, _ = w.Write([]byte("[INFO] booting\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("[INFO] ready\n"))
		flusher.Flush()
		_, _ = w.Write([]byte("[INFO] keep-running\n"))
		flusher.Flush()

		<-r.Context().Done()
		close(serverClosed)
	}))
	defer server.Close()

	expectedBody := "[INFO] booting\n[INFO] ready\n"

	tc := &models.TestCase{
		Name: "plain-stream-match-and-close",
		HTTPReq: models.HTTPReq{
			Method: "GET",
			URL:    server.URL,
		},
		HTTPResp: models.HTTPResp{
			Header: map[string]string{
				"Content-Type": "text/plain",
			},
			Body: expectedBody,
			StreamBody: []models.HTTPStreamChunk{
				{
					Data: []models.HTTPStreamDataField{
						{Key: "raw", Value: "[INFO] booting"},
					},
				},
				{
					Data: []models.HTTPStreamDataField{
						{Key: "raw", Value: "[INFO] ready"},
					},
				},
			},
		},
	}

	// Use SimulateHTTPStreaming for streaming test cases
	streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 3})
	require.NoError(t, err)
	require.NotNil(t, streamResp)

	// Compare the stream
	noiseKeys := map[string]struct{}{}
	matched, _, _, compareErr := CompareHTTPStream(tc.HTTPResp, streamResp.Reader, streamResp.StreamConfig, noiseKeys, logger)
	streamResp.Reader.Close()
	require.NoError(t, compareErr)
	require.True(t, matched)

	respBody := expectedBody
	if !matched {
		respBody = ""
	}

	assert.Equal(t, http.StatusOK, streamResp.StatusCode)
	assert.Equal(t, expectedBody, respBody)

	select {
	case <-serverClosed:
	case <-time.After(1 * time.Second):
		t.Fatal("expected plain text stream to be closed by client after matching queue")
	}
}

func TestCompareSSEFrame_DataJSONTypeMismatch_322(t *testing.T) {
	logger := zap.NewNop()
	match, reason := compareSSEFrame(
		"data:{\"value\":1}",
		"data:not-json",
		nil,
		logger,
	)

	assert.False(t, match)
	assert.Equal(t, "data-json-type mismatch", reason)
}

func TestCompareSSEFields_MultilineDataNormalization_322A(t *testing.T) {
	logger := zap.NewNop()

	expected := []sseField{
		{key: "id", value: "1", hasValue: true},
		{key: "data", value: "line-1\nline-2", hasValue: true},
	}
	actual := parseSSEFrame("id:1\ndata:line-1\ndata:line-2")

	match, reason := compareSSEFields(expected, actual, nil, logger)
	assert.True(t, match, "multiline SSE data should compare equal after normalization")
	assert.Equal(t, "", reason)
}

func TestComputeStreamingTimeoutSeconds_323(t *testing.T) {
	now := time.Now().UTC()
	tc := &models.TestCase{
		HTTPReq: models.HTTPReq{
			Timestamp: now,
		},
		HTTPResp: models.HTTPResp{
			Timestamp: now.Add(1500 * time.Millisecond),
		},
	}

	timeout := ComputeStreamingTimeoutSeconds(tc, 2)
	assert.Equal(t, uint64(12), timeout)

	preferConfigured := ComputeStreamingTimeoutSeconds(tc, 30)
	assert.Equal(t, uint64(30), preferConfigured)

	fallback := ComputeStreamingTimeoutSeconds(&models.TestCase{}, 7)
	assert.Equal(t, uint64(7), fallback)

	defaultMin := ComputeStreamingTimeoutSeconds(&models.TestCase{}, 0)
	assert.Equal(t, uint64(10), defaultMin)
}

// TestIsTime_VariousFormats_808 covers multiple scenarios for the IsTime function,
// including standard date formats (RFC3339, UnixDate), numeric timestamps as strings,
// and invalid inputs to ensure it correctly identifies time-like strings.
func TestIsTime_VariousFormats_808(t *testing.T) {
	testCases := []struct {
		name     string
		input    string
		expected bool
	}{
		{"RFC3339", "2023-01-17T16:34:58Z", true},
		{"UnixDate", "Tue Jan 17 16:34:58 UTC 2023", true},
		{"NumericTimestamp", fmt.Sprintf("%f", float64(time.Now().UnixNano())), true},
		{"AlmostNow", fmt.Sprintf("%f", float64(time.Now().Add(-time.Hour).UnixNano())), true},
		{"TooOldNumeric", fmt.Sprintf("%f", float64(time.Now().Add(-48*time.Hour).UnixNano())), false},
		{"InvalidString", "not a date", false},
		{"EmptyString", "", false},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			assert.Equal(t, tc.expected, IsTime(tc.input))
		})
	}
}

// TestToHTTPHeader_WithTimeValue_909 verifies that the ToHTTPHeader function correctly
// converts a map of strings to an http.Header object. It specifically checks that
// header values recognized as timestamps are not split by commas, while other
// comma-separated values are correctly split into slices.
func TestToHTTPHeader_WithTimeValue_909(t *testing.T) {
	// Arrange
	mockHeader := map[string]string{
		"Date":            "Tue, 17 Jan 2023 16:34:58 IST",
		"X-Custom-Header": "value1,value2",
		"Content-Type":    "application/json",
	}

	// Act
	httpHeader := ToHTTPHeader(mockHeader)

	// Assert: every YAML key replays as exactly one wire header. The
	// recorded value is kept intact (commas included) so axios-style
	// `Accept: a, b, c` does not fan out into three duplicate-name
	// wire headers — see ToHTTPHeader's comment for why fan-out
	// breaks `dict(**request.headers)`-style server middlewares.
	require.NotNil(t, httpHeader)
	assert.Equal(t, []string{"Tue, 17 Jan 2023 16:34:58 IST"}, httpHeader["Date"])
	assert.Equal(t, []string{"value1,value2"}, httpHeader["X-Custom-Header"])
	assert.Equal(t, []string{"application/json"}, httpHeader["Content-Type"])
}

// TestToHTTPHeader_AcceptListValue_FoldedSingleHeader pins the wire
// behaviour that the duplicate-Accept TypeError fix depends on: an
// axios-style `Accept: application/json, text/plain, */*` mock must
// replay as ONE wire header so server middlewares using
// `dict(**request.headers)` do not see duplicate `accept` keys.
func TestToHTTPHeader_AcceptListValue_FoldedSingleHeader(t *testing.T) {
	mockHeader := map[string]string{
		"Accept":          "application/json, text/plain, */*",
		"Accept-Encoding": "gzip, compress, deflate, br",
		"Baggage":         "trace-id=abc,user-key=def,env=test",
	}
	httpHeader := ToHTTPHeader(mockHeader)
	require.Len(t, httpHeader["Accept"], 1, "Accept must replay as one wire header")
	require.Len(t, httpHeader["Accept-Encoding"], 1, "Accept-Encoding must replay as one wire header")
	require.Len(t, httpHeader["Baggage"], 1, "Baggage must replay as one wire header")
	assert.Equal(t, "application/json, text/plain, */*", httpHeader["Accept"][0])
	assert.Equal(t, "gzip, compress, deflate, br", httpHeader["Accept-Encoding"][0])
	assert.Equal(t, "trace-id=abc,user-key=def,env=test", httpHeader["Baggage"][0])
}

// The wire lines of a recorded header, round-tripped: what ToYamlHTTPHeader and
// ToYamlHTTPHeaderLineLengths store, ToWireHTTPHeader sends back out, and only
// what they stored. A tenant header sent twice goes out twice, not folded; one
// Accept line with commas in it stays one line (the dict(**headers) fix above).
func TestToWireHTTPHeaderSendsTheRecordedLines(t *testing.T) {
	for _, tc := range []struct {
		name string
		wire http.Header
	}{
		{"a header repeated on two lines", http.Header{"X_tenant": {"acme", "acme"}}},
		{"one line carrying commas", http.Header{"Accept": {"a, b"}}},
		{"repeated lines carrying commas", http.Header{"Cookie": {"a=1,2", "b=3,,", ",c"}}},
		{"an empty line among them", http.Header{"X-Empty": {"", "x", ""}}},
		{"both together", http.Header{"X_tenant": {"acme", "acme"}, "Accept": {"a, b"}, "Host": {"h"}}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			header := ToYamlHTTPHeader(tc.wire)
			lengths := ToYamlHTTPHeaderLineLengths(tc.wire)
			for name, lines := range tc.wire {
				if _, stored := lengths[name]; stored != (len(lines) > 1) {
					t.Errorf("%s on %d line(s): lengths stored = %v", name, len(lines), stored)
				}
			}
			assert.Equal(t, tc.wire, ToWireHTTPHeader(header, lengths))
		})
	}
	assert.Nil(t, ToYamlHTTPHeaderLineLengths(http.Header{"Accept": {"a, b"}, "Host": {"h"}}),
		"a header with no repeated line records no lengths, so nothing new reaches disk")
}

// A recording made before the lengths existed replays as it always did, and a
// value that no longer is the lines it was recorded from (a template rendered,
// a secret decrypted to something else, a hand edit) is sent as one line, not
// cut at a guessed boundary.
func TestToWireHTTPHeaderFoldsWhatItCannotSplit(t *testing.T) {
	for _, tc := range []struct {
		name    string
		value   string
		lengths []int
	}{
		{"no lengths: a recording from before them", "ab,ab", nil},
		{"one length is not a repeated header", "ab,ab", []int{5}},
		{"the value grew", "ab,abb", []int{2, 2}},
		{"the value shrank", "ab,a", []int{2, 2}},
		{"no comma where a line ended", "abXab", []int{2, 2}},
		{"a line too long for the value", "ab", []int{2, 5}},
		{"a negative length", "ab,ab", []int{-1, 5}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			var lengths models.HeaderLineLengths
			if tc.lengths != nil {
				lengths = models.HeaderLineLengths{"X_tenant": tc.lengths}
			}
			got := ToWireHTTPHeader(map[string]string{"X_tenant": tc.value}, lengths)
			assert.Equal(t, http.Header{"X_tenant": {tc.value}}, got)
		})
	}
	// Lengths for a name the header no longer has add nothing.
	got := ToWireHTTPHeader(map[string]string{"Accept": "a"}, models.HeaderLineLengths{"X_tenant": {2, 2}})
	assert.Equal(t, http.Header{"Accept": {"a"}}, got)
}

// The recorded curl command reproduces the request that was recorded, a
// repeated header on its own lines included.
func TestMakeCurlCommandRepeatsARepeatedHeader(t *testing.T) {
	curl := MakeCurlCommand(models.HTTPReq{
		Method:            "GET",
		URL:               "http://localhost:8080/v1/session",
		Header:            map[string]string{"X_tenant": "acme,acme", "Accept": "a, b", "Content-Length": "0"},
		HeaderLineLengths: models.HeaderLineLengths{"X_tenant": {4, 4}},
	})
	assert.Equal(t, 2, strings.Count(curl, "--header 'X_tenant: acme' \\\n"), curl)
	assert.NotContains(t, curl, "acme,acme")
	assert.Contains(t, curl, "--header 'Accept: a, b' \\\n")
	assert.NotContains(t, curl, "Content-Length")
}

// TestParseHTTPRequest_And_Response_111 contains sub-tests for ParseHTTPRequest and
// ParseHTTPResponse, validating both success and failure cases for parsing raw
// HTTP data into their respective struct representations.
func TestParseHTTPRequest_And_Response_111(t *testing.T) {
	t.Run("ParseHTTPRequest_Valid", func(t *testing.T) {
		rawReq := "GET /test HTTP/1.1\r\nHost: example.com\r\n\r\n"
		req, err := ParseHTTPRequest([]byte(rawReq))
		require.NoError(t, err)
		assert.Equal(t, "GET", req.Method)
		assert.Equal(t, "/test", req.URL.Path)
		assert.Equal(t, "example.com", req.Host)
	})

	t.Run("ParseHTTPRequest_Invalid", func(t *testing.T) {
		rawReq := "this is not a valid request"
		_, err := ParseHTTPRequest([]byte(rawReq))
		require.Error(t, err)
	})

	t.Run("ParseHTTPResponse_Valid", func(t *testing.T) {
		rawResp := "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello, world!"
		req, _ := http.NewRequest("GET", "http://example.com", nil)
		resp, err := ParseHTTPResponse([]byte(rawResp), req)
		require.NoError(t, err)
		defer func() {
			if err := resp.Body.Close(); err != nil {
				t.Logf("failed to close response body: %v", err)
			}
		}()
		assert.Equal(t, 200, resp.StatusCode)
		body, _ := io.ReadAll(resp.Body)
		assert.Equal(t, "Hello, world!", string(body))
	})

	t.Run("ParseHTTPResponse_Invalid", func(t *testing.T) {
		rawResp := "this is not a valid response"
		req, _ := http.NewRequest("GET", "http://example.com", nil)
		_, err := ParseHTTPResponse([]byte(rawResp), req)
		require.Error(t, err)
	})
}

// TestCompressDecompress_AllEncodings_555 provides comprehensive testing for the
// Compress and Decompress functions. It checks:
// - Successful round-trip (compress then decompress) for 'gzip' and 'br' (brotli).
// - No-op behavior for unknown encodings.
// - Error handling for invalid compressed data.
func TestCompressDecompress_AllEncodings_555(t *testing.T) {
	logger := zap.NewNop()

	t.Run("Gzip", func(t *testing.T) {
		originalData := []byte("hello world, this is a test")
		compressedData, err := Compress(logger, "gzip", originalData)
		require.NoError(t, err)
		assert.NotEqual(t, originalData, compressedData)
		decompressedData, err := Decompress(logger, "gzip", compressedData, MaxDecompressedSize)
		require.NoError(t, err)
		assert.Equal(t, originalData, decompressedData)
	})

	t.Run("Brotli", func(t *testing.T) {
		originalData := []byte("hello world, this is a brotli test")
		compressedData, err := Compress(logger, "br", originalData)
		require.NoError(t, err)
		assert.NotEqual(t, originalData, compressedData)
		decompressedData, err := Decompress(logger, "br", compressedData, MaxDecompressedSize)
		require.NoError(t, err)
		assert.Equal(t, originalData, decompressedData)
	})

	t.Run("UnknownEncoding", func(t *testing.T) {
		originalData := []byte("hello world")
		compressedData, err := Compress(logger, "unknown", originalData)
		require.NoError(t, err)
		assert.Equal(t, originalData, compressedData)
		decompressedData, err := Decompress(logger, "unknown", originalData, MaxDecompressedSize)
		require.NoError(t, err)
		assert.Equal(t, originalData, decompressedData)
	})

	t.Run("DecompressError", func(t *testing.T) {
		invalidGzipData := []byte("not gzip")
		_, err := Decompress(logger, "gzip", invalidGzipData, MaxDecompressedSize)
		require.Error(t, err)

		invalidBrotliData := []byte{0xce, 0xb2, 0xcf, 0x81}
		_, err = Decompress(logger, "br", invalidBrotliData, MaxDecompressedSize)
		require.Error(t, err)
	})

	// Content-Encoding is matched case-insensitively after trimming — on
	// BOTH sides. Decompress-only normalization would store plaintext at
	// record time while Compress (exact match) replays it un-compressed
	// under a Content-Encoding header, breaking the app's client decoder.
	t.Run("EncodingNormalizedRoundTrip", func(t *testing.T) {
		originalData := []byte("normalize me")
		for _, enc := range []string{"GZIP", " gzip", "gzip ", "BR", " br "} {
			compressedData, err := Compress(logger, enc, originalData)
			require.NoError(t, err, "encoding %q", enc)
			assert.NotEqual(t, originalData, compressedData, "Compress must not pass %q through", enc)
			decompressedData, err := Decompress(logger, enc, compressedData, MaxDecompressedSize)
			require.NoError(t, err, "encoding %q", enc)
			assert.Equal(t, originalData, decompressedData, "encoding %q", enc)
		}
	})

	// "identity" (RFC 9110 §8.4.1) means no transformation — must pass
	// through unchanged, hitting the explicit identity case rather than the
	// unsupported-encoding fallback (observed via its log line, the only
	// difference between the two paths).
	t.Run("IdentityEncoding", func(t *testing.T) {
		core, logs := observer.New(zap.DebugLevel)
		obsLogger := zap.New(core)
		originalData := []byte("plain body")
		out, err := Decompress(obsLogger, "identity", originalData, MaxDecompressedSize)
		require.NoError(t, err)
		assert.Equal(t, originalData, out)
		assert.Zero(t, logs.FilterMessageSnippet("unsupported Content-Encoding").Len(),
			"identity must not hit the unsupported-encoding fallback")
		out, err = Compress(obsLogger, "identity", originalData)
		require.NoError(t, err)
		assert.Equal(t, originalData, out)
		// A genuinely unknown encoding does hit the fallback, at debug level.
		_, err = Decompress(obsLogger, "zstd", originalData, MaxDecompressedSize)
		require.NoError(t, err)
		require.Equal(t, 1, logs.FilterMessageSnippet("unsupported Content-Encoding").Len())
		assert.Equal(t, zap.DebugLevel, logs.FilterMessageSnippet("unsupported Content-Encoding").All()[0].Level)
	})

	// A payload inflating past the caller's limit must error, not OOM (#3867).
	// Uses real gzip/brotli through the public API so it fails if the capped
	// read wiring is ever reverted. The sentinel lets callers distinguish
	// oversized-but-valid from corrupt streams.
	t.Run("DecompressionBomb", func(t *testing.T) {
		const limit = 1024
		bomb := make([]byte, 8*1024) // 8 KiB of zeros: tiny compressed, inflates past limit

		gzipBomb, err := Compress(logger, "gzip", bomb)
		require.NoError(t, err)
		_, err = Decompress(logger, "gzip", gzipBomb, limit)
		require.ErrorIs(t, err, ErrDecompressedTooLarge)
		assert.Contains(t, err.Error(), "decompression bomb")

		brBomb, err := Compress(logger, "br", bomb)
		require.NoError(t, err)
		_, err = Decompress(logger, "br", brBomb, limit)
		require.ErrorIs(t, err, ErrDecompressedTooLarge)
		assert.Contains(t, err.Error(), "decompression bomb")
	})

	// A body of exactly the limit is legal; limit+1 is not. Also pins the
	// corrupt-stream case to NOT match the sentinel.
	t.Run("DecompressionLimitBoundary", func(t *testing.T) {
		const limit = 1024

		atLimit, err := Compress(logger, "gzip", make([]byte, limit))
		require.NoError(t, err)
		out, err := Decompress(logger, "gzip", atLimit, limit)
		require.NoError(t, err)
		assert.Len(t, out, limit)

		overByOne, err := Compress(logger, "gzip", make([]byte, limit+1))
		require.NoError(t, err)
		_, err = Decompress(logger, "gzip", overByOne, limit)
		require.ErrorIs(t, err, ErrDecompressedTooLarge)

		_, err = Decompress(logger, "gzip", []byte("not gzip"), limit)
		require.Error(t, err)
		assert.NotErrorIs(t, err, ErrDecompressedTooLarge)
	})
}

// TestReadAllCapped_556 exercises the capped reader directly: boundary
// sizes, tiny limits below the initial buffer, and that buffer capacity
// never overshoots limit+1 rounded up to whole 8 KiB heap pages: for
// limit+1 over 32 KiB, what the allocator gives a buffer of limit+1 bytes;
// under it, up to a page more than the size class it would give, so that
// what a decode reserves for its buffer is what it holds
// (io.ReadAll+LimitReader, which it replaced, could keep twice limit+1).
func TestReadAllCapped_556(t *testing.T) {
	t.Run("Boundaries", func(t *testing.T) {
		for _, tc := range []struct {
			name    string
			size    int64
			limit   int64
			wantErr bool
		}{
			{"empty", 0, 1024, false},
			{"under", 1023, 1024, false},
			{"exact", 1024, 1024, false},
			{"overByOne", 1025, 1024, true},
			{"zeroLimitEmpty", 0, 0, false},
			{"zeroLimitOne", 1, 0, true},
			{"aboveInitialBuffer", 100 * 1024, 200 * 1024, false},
			{"growthThenError", 300 * 1024, 200 * 1024, true},
		} {
			t.Run(tc.name, func(t *testing.T) {
				data, err := readAllCapped(bytes.NewReader(make([]byte, tc.size)), tc.limit)
				if tc.wantErr {
					require.ErrorIs(t, err, ErrDecompressedTooLarge)
					return
				}
				require.NoError(t, err)
				assert.Equal(t, tc.size, int64(len(data)))
				pages := (tc.limit + 1 + 8<<10 - 1) / (8 << 10) * (8 << 10)
				assert.LessOrEqual(t, int64(cap(data)), pages, "capacity must never overshoot limit+1 in whole pages")
			})
		}
	})

	t.Run("UnboundedStreamStops", func(t *testing.T) {
		// An infinite reader must error at the cap, not hang or OOM.
		const limit = 64 * 1024
		_, err := readAllCapped(rand.Reader, limit)
		require.ErrorIs(t, err, ErrDecompressedTooLarge)
	})
}

// TestNewCappedReader_557 covers the streaming counterpart of readAllCapped:
// exact-limit streams reach EOF, one extra byte errors, and the error is
// sticky across subsequent reads.
func TestNewCappedReader_557(t *testing.T) {
	t.Run("ExactLimitReachesEOF", func(t *testing.T) {
		r := NewCappedReader(bytes.NewReader(make([]byte, 1024)), 1024)
		data, err := io.ReadAll(r)
		require.NoError(t, err)
		assert.Len(t, data, 1024)
	})

	t.Run("OverLimitErrorsAndSticks", func(t *testing.T) {
		r := NewCappedReader(bytes.NewReader(make([]byte, 1025)), 1024)
		_, err := io.Copy(io.Discard, r)
		require.ErrorIs(t, err, ErrDecompressedTooLarge)
		// Error must persist on further reads — a consumer retrying the
		// stream must not see a clean EOF.
		_, err = r.Read(make([]byte, 1))
		require.ErrorIs(t, err, ErrDecompressedTooLarge)
	})

	t.Run("SmallChunkedReads", func(t *testing.T) {
		r := NewCappedReader(bytes.NewReader(make([]byte, 100)), 1024)
		var total int
		buf := make([]byte, 7)
		for {
			n, err := r.Read(buf)
			total += n
			if err == io.EOF {
				break
			}
			require.NoError(t, err)
		}
		assert.Equal(t, 100, total)
	})

	t.Run("UnboundedStreamStops", func(t *testing.T) {
		_, err := io.Copy(io.Discard, NewCappedReader(rand.Reader, 64*1024))
		require.ErrorIs(t, err, ErrDecompressedTooLarge)
	})
}

// zeroReader yields an endless stream of zero bytes (compresses extremely
// well — the shape of a decompression bomb).
type zeroReader struct{}

func (zeroReader) Read(p []byte) (int, error) {
	for i := range p {
		p[i] = 0
	}
	return len(p), nil
}

// TestSimulateHTTPStreaming_DecompressionBomb_558 pins the streaming replay
// wiring (#3867): a response inflating past MaxDecompressedSize must fail
// the stream read with ErrDecompressedTooLarge instead of OOMing while the
// body accumulates.
func TestSimulateHTTPStreaming_DecompressionBomb_558(t *testing.T) {
	ctx := context.Background()
	logger := zap.NewNop()
	const bombSize = MaxDecompressedSize + 64*1024

	makeBomb := func(t *testing.T, encoding string) []byte {
		t.Helper()
		var buf bytes.Buffer
		var w io.WriteCloser
		switch encoding {
		case "gzip":
			w, _ = gzip.NewWriterLevel(&buf, gzip.BestSpeed)
		case "br":
			w = brotli.NewWriterLevel(&buf, brotli.BestSpeed)
		}
		_, err := io.CopyN(w, zeroReader{}, bombSize)
		require.NoError(t, err)
		require.NoError(t, w.Close())
		return buf.Bytes()
	}

	for _, encoding := range []string{"gzip", "br"} {
		t.Run(encoding, func(t *testing.T) {
			bomb := makeBomb(t, encoding)
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
				w.Header().Set("Content-Encoding", encoding)
				w.Header().Set("Content-Type", "application/octet-stream")
				_, _ = w.Write(bomb)
			}))
			defer server.Close()

			tc := &models.TestCase{
				Name: "decompression-bomb-" + encoding,
				HTTPReq: models.HTTPReq{
					Method: "GET",
					URL:    server.URL,
					// Explicit Accept-Encoding disables the transport's
					// transparent decompression, so the body arrives
					// compressed with Content-Encoding intact — the same
					// shape SimulateHTTPStreaming handles in production.
					Header: map[string]string{"Accept-Encoding": encoding},
				},
			}

			streamResp, err := SimulateHTTPStreaming(ctx, tc, "test-set", logger, SimulationConfig{APITimeout: 10})
			require.NoError(t, err)
			require.NotNil(t, streamResp)
			defer streamResp.Reader.Close()

			_, err = io.Copy(io.Discard, streamResp.Reader)
			require.ErrorIs(t, err, ErrDecompressedTooLarge)
		})
	}
}

// TestFilterMocks_678 validates the filtering and sorting of mocks in Test and Config modes.
func TestFilterMocks_678(t *testing.T) {
	// All subtests in TestFilterMocks_678 assert LAX-mode behaviour
	// (out-of-window non-config mocks promoted to the unfiltered pool
	// rather than dropped). The env var KEPLOY_STRICT_MOCK_WINDOW is
	// parsed once at init and cached in strictWindowEnvOverride; an
	// enabling value makes filterByTimeStamp behave strict regardless
	// of the perCall flag, so our strict=false arguments no longer
	// produce the expected lax output. Skip the whole test — every
	// subtest is affected. Strict-mode behaviour is covered by
	// TestFilterByTimeStamp_StrictMode below (which runs unconditionally).
	if strictWindowEnvOverride {
		t.Skip("KEPLOY_STRICT_MOCK_WINDOW set to enabling value — lax assertions unreachable; strict coverage runs elsewhere")
	}
	logger := zap.NewNop()
	ctx := context.Background()

	now := time.Now()
	mock1 := &models.Mock{Name: "mock1", Version: "api.keploy.io/v1beta1", Spec: models.MockSpec{ReqTimestampMock: now.Add(-2 * time.Hour), ResTimestampMock: now.Add(-2 * time.Hour)}}
	mock2 := &models.Mock{Name: "mock2", Version: "api.keploy.io/v1beta1", Spec: models.MockSpec{ReqTimestampMock: now, ResTimestampMock: now}}
	mock3 := &models.Mock{Name: "mock3", Version: "api.keploy.io/v1beta1", Spec: models.MockSpec{ReqTimestampMock: now.Add(2 * time.Hour), ResTimestampMock: now.Add(2 * time.Hour)}}
	mockNoTime := &models.Mock{Name: "mockNoTime", Version: "api.keploy.io/v1beta1", Spec: models.MockSpec{}}
	mockNonKeploy := &models.Mock{Name: "mockNonKeploy", Version: "v1", Spec: models.MockSpec{ReqTimestampMock: now, ResTimestampMock: now}}

	allMocks := []*models.Mock{mock3, mock1, mock2, mockNoTime, mockNonKeploy}

	t.Run("filterByTimeStamp_NoFilters", func(t *testing.T) {
		// strict=false (default lax) — subject under test here is the
		// no-window shortcut, not the containment rule.
		filtered, unfiltered := filterByTimeStamp(ctx, logger, allMocks, time.Time{}, time.Time{}, false)
		assert.Len(t, filtered, 5)
		assert.Len(t, unfiltered, 0)
	})

	t.Run("filterByTimeStamp_ValidRange", func(t *testing.T) {
		after := now.Add(-1 * time.Hour)
		before := now.Add(1 * time.Hour)
		// Existing assertions were written under lax Option-2 semantics
		// (req-only inside window, out-of-window mocks kept as unfiltered).
		// Pass strict=false to preserve that contract; strict mode is
		// exercised separately by TestFilterByTimeStamp_StrictMode.
		filtered, unfiltered := filterByTimeStamp(ctx, logger, allMocks, after, before, false)

		// Check filtered mocks
		require.Len(t, filtered, 3)
		filteredNames := []string{}
		for _, m := range filtered {
			filteredNames = append(filteredNames, m.Name)
			assert.True(t, m.TestModeInfo.IsFiltered)
		}
		assert.ElementsMatch(t, []string{"mock2", "mockNoTime", "mockNonKeploy"}, filteredNames)

		// Check unfiltered mocks
		require.Len(t, unfiltered, 2)
		unfilteredNames := []string{}
		for _, m := range unfiltered {
			unfilteredNames = append(unfilteredNames, m.Name)
			assert.False(t, m.TestModeInfo.IsFiltered)
		}
		assert.ElementsMatch(t, []string{"mock1", "mock3"}, unfilteredNames)
	})

	t.Run("FilterTcsMocks", func(t *testing.T) {
		after := now.Add(-1 * time.Hour)
		before := now.Add(1 * time.Hour)
		// strict=false preserves the original lax expectations of this test
		// (written before Option-1 strict containment existed).
		result := FilterTcsMocks(ctx, logger, allMocks, after, before, false)
		require.Len(t, result, 3)
		assert.Equal(t, "mockNoTime", result[0].Name) // Zero time comes first
		assert.Equal(t, "mock2", result[1].Name)
		assert.Equal(t, "mockNonKeploy", result[2].Name)
	})

	t.Run("FilterConfigMocks", func(t *testing.T) {
		after := now.Add(-1 * time.Hour)
		before := now.Add(1 * time.Hour)
		// strict=false — see FilterTcsMocks note above.
		result := FilterConfigMocks(ctx, logger, allMocks, after, before, false)
		require.Len(t, result, 5)
		// Sorted filtered part
		assert.Equal(t, "mockNoTime", result[0].Name)
		assert.Equal(t, "mock2", result[1].Name)
		assert.Equal(t, "mockNonKeploy", result[2].Name)
		// Sorted unfiltered part
		assert.Equal(t, "mock1", result[3].Name)
		assert.Equal(t, "mock3", result[4].Name)
	})
}

func TestFilterConfigMocks_PrioritizesLateMockByRequestTime_3738(t *testing.T) {
	// Reproducer asserts lax-mode behaviour for out-of-window mocks;
	// skip when KEPLOY_STRICT_MOCK_WINDOW forces strict on. See
	// TestFilterMocks_678's guard for the same reason.
	if strictWindowEnvOverride {
		t.Skip("KEPLOY_STRICT_MOCK_WINDOW is set — unset it to run lax-mode assertions")
	}
	logger := zap.NewNop()
	ctx := context.Background()

	testReqTime := time.Date(2026, time.February, 12, 6, 39, 20, 766944656, time.UTC)
	testRespTime := time.Date(2026, time.February, 12, 6, 39, 20, 820398862, time.UTC)

	staleData := &models.Mock{
		Name:    "mock-62",
		Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: time.Date(2026, time.February, 12, 6, 38, 43, 559270267, time.UTC),
			ResTimestampMock: time.Date(2026, time.February, 12, 6, 38, 43, 565339949, time.UTC),
		},
	}
	countQuery := &models.Mock{
		Name:    "mock-137",
		Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: time.Date(2026, time.February, 12, 6, 39, 20, 767839595, time.UTC),
			ResTimestampMock: time.Date(2026, time.February, 12, 6, 39, 20, 809483263, time.UTC),
		},
	}
	trailingData := &models.Mock{
		Name:    "mock-138",
		Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: time.Date(2026, time.February, 12, 6, 39, 20, 809891000, time.UTC),
			ResTimestampMock: time.Date(2026, time.February, 12, 6, 39, 20, 820711000, time.UTC),
		},
	}

	// strict=false — reproducer test validates legacy Option-2 behaviour
	// (trailing response kept in filtered even though it landed a
	// microsecond past beforeTime).
	filtered, unfiltered := filterByTimeStamp(ctx, logger, []*models.Mock{trailingData, staleData, countQuery}, testReqTime, testRespTime, false)
	require.Len(t, filtered, 2)
	assert.ElementsMatch(t, []string{"mock-137", "mock-138"}, []string{filtered[0].Name, filtered[1].Name})
	require.Len(t, unfiltered, 1)
	assert.Equal(t, "mock-62", unfiltered[0].Name)

	result := FilterConfigMocks(ctx, logger, []*models.Mock{trailingData, staleData, countQuery}, testReqTime, testRespTime, false)
	require.Len(t, result, 3)
	assert.Equal(t, "mock-137", result[0].Name)
	assert.Equal(t, "mock-138", result[1].Name)
	assert.Equal(t, "mock-62", result[2].Name)
}

// TestFilterByTimeStamp_StrictMode validates the strict-window opt-in.
// The in-window predicate is REQUEST-timestamp only (responses may
// legitimately straggle past beforeTime), so strict and lax both keep
// mocks whose req is inside [after,before]. The behavioural difference
// is SOLELY out-of-window handling: strict DROPS non-config mocks that
// would leak into the unfiltered pool, while lax preserves the legacy
// cross-test config pool promotion. Config-tagged mocks survive in
// both modes.
func TestFilterByTimeStamp_StrictMode(t *testing.T) {
	// The process-wide env value (strictWindowEnvOverride /
	// strictWindowEnvExplicitOff) is parsed once at init and then
	// cached — we can't os.Setenv() it mid-test. Per-subtest skips
	// below keep each assertion running when its mode is reachable
	// and skip only the ones that can't.
	logger := zap.NewNop()
	ctx := context.Background()

	now := time.Now()
	after := now.Add(-1 * time.Hour)
	before := now.Add(1 * time.Hour)

	inWindow := &models.Mock{
		Name: "in-window-data", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{ReqTimestampMock: now, ResTimestampMock: now.Add(time.Second)},
	}
	straggler := &models.Mock{
		// Req inside window but Res past beforeTime — under the req-only
		// containment rule, BOTH lax and strict keep it in the filtered
		// set (response straggle is normal async completion, not bleed).
		Name: "straggler-data", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{ReqTimestampMock: now, ResTimestampMock: now.Add(2 * time.Hour)},
	}
	stale := &models.Mock{
		// Both endpoints before window — both modes drop from filteredMocks
		// (lax pushes to unfiltered, strict drops entirely).
		Name: "stale-data", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{ReqTimestampMock: now.Add(-3 * time.Hour), ResTimestampMock: now.Add(-3 * time.Hour)},
	}
	staleConfig := &models.Mock{
		// Stale but type=config — strict keeps in unfiltered pool, lax also keeps.
		Name: "stale-config", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: now.Add(-3 * time.Hour),
			ResTimestampMock: now.Add(-3 * time.Hour),
			Metadata:         map[string]string{"type": "config"},
		},
	}
	all := []*models.Mock{inWindow, straggler, stale, staleConfig}

	t.Run("default lax promotes stale-data to unfiltered (legacy leak)", func(t *testing.T) {
		// strict=false (default). Req-in-window rule for the filtered
		// partition; out-of-window non-config mocks get promoted to the
		// unfiltered pool (the legacy bleed). Env forcing strict on
		// defeats the lax path — skip this subtest only.
		if strictWindowEnvOverride {
			t.Skip("KEPLOY_STRICT_MOCK_WINDOW forces strict on — lax semantics not reachable")
		}
		filtered, unfiltered := filterByTimeStamp(ctx, logger, all, after, before, false)
		filteredNames := mockNames(filtered)
		unfilteredNames := mockNames(unfiltered)
		assert.ElementsMatch(t, []string{"in-window-data", "straggler-data"}, filteredNames)
		assert.ElementsMatch(t, []string{"stale-data", "stale-config"}, unfilteredNames)
	})

	t.Run("opt-in strict keeps straggler (req in window); drops stale-data; keeps stale-config", func(t *testing.T) {
		// strict=true — opt into Option-1 containment. Env forcing
		// strict OFF defeats the strict path — skip this subtest only.
		if strictWindowEnvExplicitOff {
			t.Skip("KEPLOY_STRICT_MOCK_WINDOW=0 forces strict off — strict semantics not reachable")
		}
		filtered, unfiltered := filterByTimeStamp(ctx, logger, all, after, before, true)
		filteredNames := mockNames(filtered)
		unfilteredNames := mockNames(unfiltered)
		assert.ElementsMatch(t, []string{"in-window-data", "straggler-data"}, filteredNames)
		assert.ElementsMatch(t, []string{"stale-config"}, unfilteredNames)
	})
}

// Locks in the contract that a mock whose response timestamp precedes
// its request timestamp is DROPPED from both filtered and unfiltered
// partitions, in BOTH strict and lax modes — the response-before-
// request state is always an inconsistent recording (clock skew,
// serialisation bug, or file corruption) and keeping it in either pool
// risks confusing the matcher's scoring. See filterByTimeStamp's
// "defensive sanity check" in pkg/util.go.
func TestFilterByTimeStamp_DropsInvalidTimestampOrder(t *testing.T) {
	logger := zap.NewNop()
	ctx := context.Background()

	now := time.Now()
	after := now.Add(-1 * time.Hour)
	before := now.Add(1 * time.Hour)

	// Req inside window, but Res BEFORE Req by 1s — invalid order.
	invalid := &models.Mock{
		Name: "invalid-order", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: now,
			ResTimestampMock: now.Add(-time.Second),
		},
	}
	// Same shape but type=config — still dropped; invalid-order check
	// fires before the config-vs-data partition.
	invalidConfig := &models.Mock{
		Name: "invalid-order-config", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: now,
			ResTimestampMock: now.Add(-time.Second),
			Metadata:         map[string]string{"type": "config"},
		},
	}
	// Sanity: a well-ordered in-window mock survives alongside drops.
	valid := &models.Mock{
		Name: "valid-in-window", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: now,
			ResTimestampMock: now.Add(time.Second),
		},
	}
	all := []*models.Mock{invalid, invalidConfig, valid}

	t.Run("lax mode drops invalid-order from BOTH partitions", func(t *testing.T) {
		filtered, unfiltered := filterByTimeStamp(ctx, logger, all, after, before, false)
		assert.ElementsMatch(t, []string{"valid-in-window"}, mockNames(filtered))
		assert.Empty(t, mockNames(unfiltered))
	})

	t.Run("strict mode drops invalid-order from BOTH partitions", func(t *testing.T) {
		filtered, unfiltered := filterByTimeStamp(ctx, logger, all, after, before, true)
		assert.ElementsMatch(t, []string{"valid-in-window"}, mockNames(filtered))
		assert.Empty(t, mockNames(unfiltered))
	})
}

func mockNames(ms []*models.Mock) []string {
	out := make([]string, 0, len(ms))
	for _, m := range ms {
		out = append(out, m.Name)
	}
	return out
}

// TestFilterByTimeStampTierAware_StartupPreservation pins the
// tier-aware strictMockWindow contract. Given the five cases from the
// contract (startup, in-window, stale prev-test, session, connection)
// plus the stale-past-window case, the filter KEEPs startup + session +
// connection + in-window and DROPs only genuinely stale cross-test
// per-test mocks. See pkg/util.go's filterByTimeStampTierAware doc.
func TestFilterByTimeStampTierAware_StartupPreservation(t *testing.T) {
	if strictWindowEnvExplicitOff {
		t.Skip("KEPLOY_STRICT_MOCK_WINDOW=0 forces strict off — strict-tier semantics not reachable")
	}
	logger := zap.NewNop()
	ctx := context.Background()

	// firstWindowStart < previousTestEnd < currentStart < currentEnd
	//         |             |                 |              |
	//      2:00 UTC      2:30 UTC          3:00 UTC       3:10 UTC
	firstWindowStart := time.Date(2026, 4, 22, 2, 0, 0, 0, time.UTC)
	previousTestEnd := time.Date(2026, 4, 22, 2, 30, 0, 0, time.UTC)
	currentStart := time.Date(2026, 4, 22, 3, 0, 0, 0, time.UTC)
	currentEnd := time.Date(2026, 4, 22, 3, 10, 0, 0, time.UTC)

	// startup: per-test mock recorded BEFORE any test fired (req <
	// firstWindowStart). Legacy blanket-drop killed this; tier-aware
	// must preserve.
	startup := &models.Mock{
		Name: "startup-pertest", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: firstWindowStart.Add(-5 * time.Minute),
			ResTimestampMock: firstWindowStart.Add(-5*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}
	// inWindow: per-test mock whose req is inside [currentStart, currentEnd].
	inWindow := &models.Mock{
		Name: "in-window-pertest", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: currentStart.Add(time.Minute),
			ResTimestampMock: currentStart.Add(time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}
	// stalePrev: per-test mock from a previous test window
	// (firstWindowStart <= req < currentStart). REAL bleed — must drop.
	stalePrev := &models.Mock{
		Name: "stale-prev-test", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: previousTestEnd.Add(-5 * time.Minute),
			ResTimestampMock: previousTestEnd.Add(-5*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}
	// session: LifetimeSession is always preserved (spans tests).
	session := &models.Mock{
		Name: "session-scoped", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: previousTestEnd.Add(-10 * time.Minute),
			ResTimestampMock: previousTestEnd.Add(-10*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimeSession},
	}
	// connection: LifetimeConnection is always preserved (conn-bounded).
	connection := &models.Mock{
		Name: "connection-scoped", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: previousTestEnd.Add(-15 * time.Minute),
			ResTimestampMock: previousTestEnd.Add(-15*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimeConnection},
	}
	// stalePast: per-test mock whose req is PAST currentEnd (future-test
	// bleed). Also genuine stale and must be dropped.
	stalePast := &models.Mock{
		Name: "stale-past-test", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: currentEnd.Add(10 * time.Minute),
			ResTimestampMock: currentEnd.Add(10*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}

	all := []*models.Mock{startup, inWindow, stalePrev, session, connection, stalePast}

	filtered, unfiltered := filterByTimeStampTierAware(ctx, logger, all, currentStart, currentEnd, true, firstWindowStart)

	// The startup per-test mock survives in `filtered` so MockManager's
	// SetMocksWithWindow partition-by-firstWindowStart routes it into
	// the startup tree. The in-window per-test mock also lives in
	// `filtered` — that's its current tier.
	assert.ElementsMatch(t, []string{"startup-pertest", "in-window-pertest"}, mockNames(filtered),
		"startup + in-window per-test mocks must survive in filtered slice")

	// Session and connection mocks go to unfiltered (reusable-across-
	// tests pool). Stale previous-test and stale future-test per-test
	// mocks are DROPPED — genuine cross-test bleed.
	assert.ElementsMatch(t, []string{"session-scoped", "connection-scoped"}, mockNames(unfiltered),
		"session + connection must survive in unfiltered slice; stale per-test must drop")
}

// TestFilterByTimeStampTierAware_ZeroFirstStart_Fallback pins the
// fallback contract: when firstWindowStart is zero (no test has fired
// yet on the underlying MockManager, or the caller couldn't thread
// it), the tier-aware filter reverts to the legacy blanket-drop
// semantics so behaviour is strictly no worse than before the fix.
func TestFilterByTimeStampTierAware_ZeroFirstStart_Fallback(t *testing.T) {
	if strictWindowEnvExplicitOff {
		t.Skip("KEPLOY_STRICT_MOCK_WINDOW=0 forces strict off — strict fallback not reachable")
	}
	logger := zap.NewNop()
	ctx := context.Background()

	now := time.Now()
	after := now.Add(-1 * time.Hour)
	before := now.Add(1 * time.Hour)

	// Per-test mock predating the current window. Under tier-aware with
	// a known firstWindowStart this would be preserved as startup; under
	// zero-firstWindowStart fallback it behaves like the legacy blanket
	// drop (dropped under strict).
	preWindow := &models.Mock{
		Name: "pre-window", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: now.Add(-3 * time.Hour),
			ResTimestampMock: now.Add(-3*time.Hour + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}
	inWindow := &models.Mock{
		Name: "in-window", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: now,
			ResTimestampMock: now.Add(time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}

	all := []*models.Mock{preWindow, inWindow}

	filtered, unfiltered := filterByTimeStampTierAware(ctx, logger, all, after, before, true, time.Time{})

	// Zero firstWindowStart → legacy blanket-drop — pre-window dropped,
	// in-window kept.
	assert.ElementsMatch(t, []string{"in-window"}, mockNames(filtered),
		"zero firstWindowStart must fall back to legacy blanket drop for per-test mocks")
	assert.Empty(t, mockNames(unfiltered),
		"zero firstWindowStart strict mode must leave per-test unfiltered empty")
}

// TestFilterByTimeStamp_ScopeMetadataIgnored_LifetimeAndTimestampAuthoritative
// pins the mockdb/filter-layer authoritative per-test router contract:
// pool routing uses ONLY Lifetime + ReqTimestampMock. The recorder-
// stamped metadata["scope"] string is informational on the wire and
// MUST NOT influence routing. This regression-guards the debug-bundle
// symptom where perTest-lifetime mocks with a recorder-stamped
// scope=session (context-probe captured between test-window
// transitions) were incorrectly dropped into the session pool.
//
// Contract:
//
//  1. mock{Lifetime=perTest, metadata.scope=session, req ∈ window} →
//     per-test pool (filtered). The timestamp attribution wins; scope
//     is ignored.
//  2. mock{Lifetime=session, metadata.scope=test:N, req ∈ window} →
//     session pool (unfiltered). Lifetime-first short-circuit wins;
//     neither scope nor the fact that req happens to land inside the
//     current test's window matters for the session-lifetime class.
//  3. mock{Lifetime=connection, metadata.scope=session, req ∈ window} →
//     session pool (unfiltered). Same lifetime-first semantics as (2).
func TestFilterByTimeStamp_ScopeMetadataIgnored_LifetimeAndTimestampAuthoritative(t *testing.T) {
	if strictWindowEnvExplicitOff {
		t.Skip("KEPLOY_STRICT_MOCK_WINDOW=0 forces strict off — lifetime-first routing only exercised under strict")
	}
	logger := zap.NewNop()
	ctx := context.Background()

	start := time.Date(2026, 4, 23, 10, 0, 0, 0, time.UTC)
	end := start.Add(10 * time.Second)
	reqInWindow := start.Add(time.Second)
	resInWindow := reqInWindow.Add(time.Millisecond)

	// Case 1: the user's exact anomaly shape — a perTest-lifetime mock
	// with recorder-stamped scope=session, timestamp inside the window.
	// Must land in the per-test (filtered) pool via timestamp
	// attribution; the scope tag is noise.
	perTestWithSessionScope := &models.Mock{
		Name:    "pertest-scope-session-anomaly",
		Version: "api.keploy.io/v1beta1",
		Kind:    models.PostgresV3,
		Spec: models.MockSpec{
			ReqTimestampMock: reqInWindow,
			ResTimestampMock: resInWindow,
			// The recorder stamped scope=session (context-probe
			// thought it was between tests) AND type=mocks (semantic
			// class classified the query as per-test). Exactly the
			// debug-bundle's "lifetime: perTest, scope: session" shape.
			Metadata: map[string]string{"type": "mocks", "scope": "session"},
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}
	// Case 2: a session-lifetime mock stamped scope=test:N, req in
	// window. Lifetime-first short-circuit wins — session pool.
	sessionWithTestScope := &models.Mock{
		Name:    "session-scope-test-anomaly",
		Version: "api.keploy.io/v1beta1",
		Kind:    models.PostgresV3,
		Spec: models.MockSpec{
			ReqTimestampMock: reqInWindow,
			ResTimestampMock: resInWindow,
			Metadata:         map[string]string{"type": "config", "scope": "test:sap-customer360"},
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimeSession},
	}
	// Case 3: a connection-lifetime mock stamped scope=session, req in
	// window. Lifetime-first short-circuit wins — session/unfiltered
	// pool (MockManager seeds the per-connID pool from unfiltered).
	connWithSessionScope := &models.Mock{
		Name:    "connection-scope-session",
		Version: "api.keploy.io/v1beta1",
		Kind:    models.PostgresV3,
		Spec: models.MockSpec{
			ReqTimestampMock: reqInWindow,
			ResTimestampMock: resInWindow,
			Metadata:         map[string]string{"type": "connection", "scope": "session", "connID": "pg-conn-1"},
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimeConnection},
	}
	// Case 4 (baseline sanity): an ordinary perTest-lifetime mock with
	// no scope tag at all, req in window. Must still land in per-test.
	// Guards against a regression where the lifetime-first branch
	// accidentally swallows perTest traffic.
	perTestNoScope := &models.Mock{
		Name:    "pertest-no-scope",
		Version: "api.keploy.io/v1beta1",
		Kind:    models.PostgresV3,
		Spec: models.MockSpec{
			ReqTimestampMock: reqInWindow,
			ResTimestampMock: resInWindow,
			Metadata:         map[string]string{"type": "mocks"},
		},
		TestModeInfo: models.TestModeInfo{Lifetime: models.LifetimePerTest},
	}

	all := []*models.Mock{perTestWithSessionScope, sessionWithTestScope, connWithSessionScope, perTestNoScope}

	filtered, unfiltered := filterByTimeStampTierAware(ctx, logger, all, start, end, true, start)

	assert.ElementsMatch(t,
		[]string{"pertest-scope-session-anomaly", "pertest-no-scope"},
		mockNames(filtered),
		"perTest-lifetime mocks in window must land in the per-test pool irrespective of metadata[\"scope\"]")
	assert.ElementsMatch(t,
		[]string{"session-scope-test-anomaly", "connection-scope-session"},
		mockNames(unfiltered),
		"Session/Connection-lifetime mocks must land in the session pool irrespective of metadata[\"scope\"] or whether req sits inside the current window")
}

// TestHasExplicitPort_IPv6_777 validates the hasExplicitPort function with various host strings,
// including IPv4, IPv6, and hostnames, both with and without ports.
func TestHasExplicitPort_IPv6_777(t *testing.T) {
	testCases := []struct {
		name     string
		host     string
		expected bool
	}{
		{"IPv4WithPort", "127.0.0.1:8080", true},
		{"IPv4WithoutPort", "127.0.0.1", false},
		{"IPv6WithPort", "[::1]:8080", true},
		{"IPv6WithoutPort", "[::1]", false},
		{"IPv6WithoutBrackets", "::1", false}, // Invalid for SplitHostPort, so false
		{"HostnameWithPort", "localhost:8080", true},
		{"HostnameWithoutPort", "localhost", false},
		{"InvalidPort", "localhost:http", false},    // Non-numeric port
		{"FullURL", "http://localhost:8080", false}, // SplitHostPort fails on scheme
		{"IPv6ComplexWithPort", "[2001:db8::1]:8080", true},
		{"IPv6ComplexWithoutPort", "[2001:db8::1]", false},
		{"ColonInPath", "localhost:8080/foo", false}, // SplitHostPort fails
	}
	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			assert.Equal(t, tc.expected, hasExplicitPort(tc.host))
		})
	}
}

// A URL that already says where the app is (test.basePath) is where the
// request goes under TargetFromURL, and under that address's name. The
// recorded app_port otherwise replaces the URL's port, and the recorded Host
// header the URL's host: the test goes to the port the app was recorded on,
// asking for the app by the name it had there.
func TestSimulateHTTP_TargetFromURLLeavesTheRecordedAddressOut(t *testing.T) {
	// askedAs gets the Host header of each request the server answers.
	askedAs := make(chan string, 1)
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		askedAs <- r.Host
		w.WriteHeader(http.StatusOK)
	}))
	defer server.Close()
	serverHost := strings.TrimPrefix(server.URL, "http://")

	// The port the test was recorded on: nothing listens there any more.
	recordedPort := refusedPort(t)

	newCase := func(appPort uint16) *models.TestCase {
		return &models.TestCase{
			Name:    "get-orders",
			AppPort: appPort,
			HTTPReq: models.HTTPReq{
				Method: "GET",
				URL:    server.URL + "/orders",
				Header: map[string]string{"Host": "recorded.internal:8080"},
			},
		}
	}

	resp, err := SimulateHTTP(context.Background(), newCase(recordedPort), "test-set", zap.NewNop(),
		SimulationConfig{APITimeout: 5, TargetFromURL: true})
	require.NoError(t, err, "the request must go to the URL's own port")
	assert.Equal(t, http.StatusOK, resp.StatusCode)
	assert.Equal(t, serverHost, <-askedAs, "the app must be asked for by the URL's host")

	// Without it the recorded port still wins, as it must for an app keploy
	// starts, and so does the recorded name.
	_, err = SimulateHTTP(context.Background(), newCase(recordedPort), "test-set", zap.NewNop(),
		SimulationConfig{APITimeout: 5})
	require.Error(t, err, "the recorded app_port did not replace the URL's port")
	assert.Contains(t, err.Error(), fmt.Sprintf(":%d", recordedPort))

	_, err = SimulateHTTP(context.Background(), newCase(0), "test-set", zap.NewNop(),
		SimulationConfig{APITimeout: 5})
	require.NoError(t, err)
	assert.Equal(t, "recorded.internal:8080", <-askedAs, "the recorded Host header was not sent")
}

// Under TargetFromURL the app is asked for by the URL's host as a client
// writes it: without the port when it is the scheme's own. ResolveTestTarget
// writes that port into every URL, and left to Go the request asked for
// "staging.example.com:443", which a front that matches the name as written
// does not route to the app every client reaches as "staging.example.com".
func TestPrepareHTTPRequest_TargetFromURLAsksForTheHostAsAClientWritesIt(t *testing.T) {
	for _, tc := range []struct {
		name, url  string
		configPort uint32
		want       string
	}{
		{"http and no port", "http://staging.example.com/orders", 0, "staging.example.com"},
		{"https and no port", "https://staging.example.com/orders", 0, "staging.example.com"},
		{"the scheme's own port written out", "https://staging.example.com:443/orders", 0, "staging.example.com"},
		{"IPv6 and no port", "http://[::1]/orders", 0, "[::1]"},
		{"a port of its own", "http://staging.example.com:8080/orders", 0, "staging.example.com:8080"},
		{"the other scheme's port", "http://staging.example.com:443/orders", 0, "staging.example.com:443"},
		{"a port set for the tests", "http://staging.example.com/orders", 8080, "staging.example.com:8080"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			test := &models.TestCase{
				Name:    "get-orders",
				AppPort: 9090,
				HTTPReq: models.HTTPReq{
					Method: "GET",
					URL:    tc.url,
					Header: map[string]string{"Host": "recorded.internal:9090"},
				},
			}
			prepared, err := prepareHTTPRequest(context.Background(), test, "test-set", zap.NewNop(),
				SimulationConfig{TargetFromURL: true, ConfigPort: tc.configPort})
			require.NoError(t, err)

			// The request as it goes on the wire.
			var wire bytes.Buffer
			require.NoError(t, prepared.Request.Write(&wire))
			assert.Contains(t, wire.String(), "\r\nHost: "+tc.want+"\r\n")
		})
	}
}

func TestResolveTestTarget(t *testing.T) {
	logger := zap.NewNop()

	tests := []struct {
		name            string
		originalTarget  string
		urlReplacements map[string]string
		portMappings    map[uint32]uint32
		configHost      string
		appPort         uint16
		configPort      uint32
		isHTTP          bool
		expectedTarget  string
		expectError     bool
	}{
		// HTTP Scenarios
		{
			name:           "HTTP_NoOverrides",
			originalTarget: "http://example.com/api",
			isHTTP:         true,
			expectedTarget: "http://example.com:80/api",
		},
		{
			name:           "HTTP_AppPortOverride",
			originalTarget: "http://example.com/api",
			appPort:        8080,
			isHTTP:         true,
			expectedTarget: "http://example.com:8080/api",
		},
		{
			name:           "HTTP_ConfigPortOverride",
			originalTarget: "http://example.com/api",
			appPort:        8080,
			configPort:     9090,
			isHTTP:         true,
			expectedTarget: "http://example.com:9090/api",
		},
		{
			name:           "HTTP_ConfigHostOverride",
			originalTarget: "http://example.com/api",
			configHost:     "localhost",
			isHTTP:         true,
			expectedTarget: "http://localhost:80/api",
		},
		{
			name:            "HTTP_ReplacementWithPort_Final",
			originalTarget:  "http://example.com/api",
			urlReplacements: map[string]string{"example.com": "localhost:3000"},
			configPort:      9090, // Should be ignored
			isHTTP:          true,
			expectedTarget:  "http://localhost:3000/api",
		},
		{
			name:            "HTTP_ReplacementWithoutPort_AppliesOverrides",
			originalTarget:  "http://example.com/api",
			urlReplacements: map[string]string{"example.com": "new-host"},
			configPort:      9090,
			isHTTP:          true,
			expectedTarget:  "http://new-host:9090/api",
		},
		{
			name:            "HTTP_PortMappingOverridesReplacementPort",
			originalTarget:  "http://example.com/api",
			urlReplacements: map[string]string{"example.com": "localhost:3000"},
			portMappings:    map[uint32]uint32{3000: 4000},
			configPort:      9090, // Still ignored by replacement-with-port before mapping.
			isHTTP:          true,
			expectedTarget:  "http://localhost:4000/api",
		},
		{
			name:           "HTTP_PortMappingOverridesConfigAndAppPort",
			originalTarget: "http://example.com/api",
			appPort:        8080,
			configPort:     9090,
			portMappings:   map[uint32]uint32{9090: 7070},
			isHTTP:         true,
			expectedTarget: "http://example.com:7070/api",
		},
		{
			name:           "HTTPS_DefaultPort",
			originalTarget: "https://example.com/api",
			isHTTP:         true,
			expectedTarget: "https://example.com:443/api",
		},

		// gRPC Scenarios
		{
			name:           "GRPC_NoOverrides",
			originalTarget: "example.com",
			isHTTP:         false,
			expectedTarget: "example.com:443",
		},
		{
			name:           "GRPC_ExistingPort_NoOverrides",
			originalTarget: "example.com:50051",
			isHTTP:         false,
			expectedTarget: "example.com:50051",
		},
		{
			name:           "GRPC_AppPortOverride",
			originalTarget: "example.com",
			appPort:        8080,
			isHTTP:         false,
			expectedTarget: "example.com:8080",
		},
		{
			name:           "GRPC_ConfigPortOverride",
			originalTarget: "example.com",
			configPort:     9090,
			isHTTP:         false,
			expectedTarget: "example.com:9090",
		},
		{
			name:           "GRPC_ConfigHostOverride",
			originalTarget: "example.com:50051",
			configHost:     "localhost",
			isHTTP:         false,
			expectedTarget: "localhost:50051",
		},
		{
			name:            "GRPC_ReplacementWithPort_Final",
			originalTarget:  "example.com:50051",
			urlReplacements: map[string]string{"example.com": "localhost:3000"},
			configPort:      9090, // Should be ignored
			isHTTP:          false,
			expectedTarget:  "localhost:3000:50051", // Replacement is literal substitution first
		},
		{
			name:            "GRPC_ReplacementWithPort_ExactMatch",
			originalTarget:  "example.com",
			urlReplacements: map[string]string{"example.com": "localhost:3000"},
			configPort:      9090,
			isHTTP:          false,
			expectedTarget:  "localhost:3000",
		},
		{
			name:           "GRPC_IPv6_Host",
			originalTarget: "[::1]:50051",
			configPort:     9090,
			isHTTP:         false,
			expectedTarget: "[::1]:9090",
		},
		{
			name:           "GRPC_PortMappingOverridesConfigPort",
			originalTarget: "example.com:50051",
			configPort:     50052,
			portMappings:   map[uint32]uint32{50052: 50053},
			isHTTP:         false,
			expectedTarget: "example.com:50053",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := ResolveTestTarget(tt.originalTarget, tt.urlReplacements, tt.portMappings, tt.configHost, tt.appPort, tt.configPort, tt.isHTTP, logger)
			if tt.expectError {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.expectedTarget, got)
			}
		})
	}
}

func TestResolveTestTarget_EdgeCases(t *testing.T) {
	logger := zap.NewNop()

	t.Run("HTTP_InvalidURL", func(t *testing.T) {
		_, err := ResolveTestTarget("http://[::1]:namedport", nil, nil, "", 0, 0, true, logger)
		assert.Error(t, err)
	})

	t.Run("GRPC_ConfigHost_ReplaceError", func(t *testing.T) {
		// Mock logic or ensure specific error condition if possible, though ReplaceGrpcHost is robust
	})
}

// TestFilterByTimeStampThreeTier validates the Lifetime-elevated three-
// pool partition. Contract:
//   - LifetimeConnection           → startupMocks (connection-scoped reuse)
//   - LifetimeSession              → unfilteredMocks (session-scoped reuse)
//   - LifetimePerTest in-window    → filteredMocks
//   - legacy metadata.type=config  → unfilteredMocks (fallback)
//   - legacy metadata.type=connection → startupMocks (fallback)
//   - legacy no-type per-test      → filteredMocks (legacy fallback path)
//   - startup-init (req < firstWindowStart) under strict → startupMocks
func TestFilterByTimeStampThreeTier(t *testing.T) {
	if strictWindowEnvExplicitOff {
		t.Skip("KEPLOY_STRICT_MOCK_WINDOW=0 forces strict off — strict-tier semantics not reachable")
	}
	logger := zap.NewNop()
	ctx := context.Background()

	firstWindowStart := time.Date(2026, 4, 22, 2, 0, 0, 0, time.UTC)
	currentStart := time.Date(2026, 4, 22, 3, 0, 0, 0, time.UTC)
	currentEnd := time.Date(2026, 4, 22, 3, 10, 0, 0, time.UTC)

	// In-window PerTest → filtered.
	inWindow := &models.Mock{
		Name: "perTest-in", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: currentStart.Add(time.Minute),
			ResTimestampMock: currentStart.Add(time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{
			Lifetime:        models.LifetimePerTest,
			LifetimeDerived: true,
		},
	}
	// Session → unfiltered.
	session := &models.Mock{
		Name: "session", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: firstWindowStart.Add(-10 * time.Minute),
			ResTimestampMock: firstWindowStart.Add(-10*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{
			Lifetime:        models.LifetimeSession,
			LifetimeDerived: true,
		},
	}
	// Connection → startup.
	conn := &models.Mock{
		Name: "conn", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: firstWindowStart.Add(-5 * time.Minute),
			ResTimestampMock: firstWindowStart.Add(-5*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{
			Lifetime:        models.LifetimeConnection,
			LifetimeDerived: true,
		},
	}
	// Startup-init PerTest (req < firstWindowStart) → startup.
	startupInit := &models.Mock{
		Name: "startup-init", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			ReqTimestampMock: firstWindowStart.Add(-2 * time.Minute),
			ResTimestampMock: firstWindowStart.Add(-2*time.Minute + time.Millisecond),
		},
		TestModeInfo: models.TestModeInfo{
			Lifetime:        models.LifetimePerTest,
			LifetimeDerived: true,
		},
	}
	// Legacy config (no Lifetime, just metadata.type=config) → unfiltered.
	legacyConfig := &models.Mock{
		Name: "legacy-config", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			Metadata:         map[string]string{"type": "config"},
			ReqTimestampMock: firstWindowStart.Add(-20 * time.Minute),
			ResTimestampMock: firstWindowStart.Add(-20*time.Minute + time.Millisecond),
		},
	}
	// Legacy connection (no Lifetime, just metadata.type=connection) → startup.
	legacyConn := &models.Mock{
		Name: "legacy-conn", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			Metadata:         map[string]string{"type": "connection"},
			ReqTimestampMock: firstWindowStart.Add(-25 * time.Minute),
			ResTimestampMock: firstWindowStart.Add(-25*time.Minute + time.Millisecond),
		},
	}
	// Legacy per-test (no Lifetime, metadata.type="mocks") — treated as PerTest.
	legacyPerTest := &models.Mock{
		Name: "legacy-pertest", Version: "api.keploy.io/v1beta1",
		Spec: models.MockSpec{
			Metadata:         map[string]string{"type": "mocks"},
			ReqTimestampMock: currentStart.Add(2 * time.Minute),
			ResTimestampMock: currentStart.Add(2*time.Minute + time.Millisecond),
		},
	}

	all := []*models.Mock{inWindow, session, conn, startupInit, legacyConfig, legacyConn, legacyPerTest}

	filtered, unfiltered, startup := FilterByTimeStampThreeTier(ctx, logger, all, currentStart, currentEnd, true, firstWindowStart)

	// Helper to collect names.
	names := func(ms []*models.Mock) []string {
		out := make([]string, 0, len(ms))
		for _, m := range ms {
			out = append(out, m.Name)
		}
		return out
	}

	assert.ElementsMatch(t, []string{"perTest-in", "legacy-pertest"}, names(filtered),
		"perTest in-window mocks land in filtered; legacy-untyped perTest falls through")
	assert.ElementsMatch(t, []string{"session", "legacy-config"}, names(unfiltered),
		"session-lifetime + legacy config-tagged land in unfiltered")
	assert.ElementsMatch(t, []string{"conn", "startup-init", "legacy-conn"}, names(startup),
		"connection-lifetime + startup-init PerTest + legacy connection-tagged land in startup")
}

// AgentHealthTicker must find an agent whose health answer takes longer than
// its first check waits. Reached through a Kubernetes port-forward over a slow
// link, every new connection costs several round trips before the request is
// even sent (two stream set-ups, then the GET): about 735ms at a 245ms round
// trip. A fixed 500ms per check cut every one of them off, so the agent was
// never found ready, however long the caller waited.
func TestAgentHealthTicker_ReachesAnAgentSlowerThanTheFirstCheck(t *testing.T) {
	resetProbeOnce(t)

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/agent/health" {
			w.WriteHeader(http.StatusNotFound)
			return
		}
		time.Sleep(750 * time.Millisecond)
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte(`"OK"`))
	}))
	t.Cleanup(srv.Close)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	readyCh := make(chan bool, 1)
	go AgentHealthTicker(ctx, zap.NewNop(), srv.URL+"/agent", readyCh)

	ready, open := <-readyCh
	require.True(t, open && ready, "the agent answered every health check in 750ms and was never found ready")
}

// A health check's headers are the agent's only when it answered 200: an
// answer other than 200 (a proxy's 503, an agent still starting) says nothing
// about the agent, whatever headers it carries, and is no answer at all, as
// none is when nothing listens. Each check here has a definite answer, so none
// has a deadline.
func TestAgentHealthAnswer_IsOnlyAHealthyAnswersHeaders(t *testing.T) {
	answer := func(status int) string {
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.Header().Set("X-Keploy-Proven", "off")
			w.WriteHeader(status)
		}))
		t.Cleanup(srv.Close)
		return srv.URL + "/agent"
	}

	header := agentHealthAnswer(context.Background(), zap.NewNop(), &http.Client{}, answer(http.StatusOK))
	require.NotNil(t, header, "a 200 is a healthy answer")
	require.Equal(t, "off", header.Get("X-Keploy-Proven"), "the headers returned are not the answer's")

	require.Nil(t, agentHealthAnswer(context.Background(), zap.NewNop(), &http.Client{}, answer(http.StatusServiceUnavailable)),
		"the headers of a 503 were taken for the agent's")

	goneURL := "http://" + testnet.RefusedAddr(t, "tcp") // nothing answers there, nor can come to
	require.Nil(t, agentHealthAnswer(context.Background(), zap.NewNop(), &http.Client{}, goneURL+"/agent"),
		"an address nothing answers on gave headers")
}

// Each health check waits twice as long as the last one that ran out of time,
// up to a ceiling; a check that failed fast (nothing listening yet) or got an
// answer leaves the wait as it was, so a local agent keeps its quick check.
func TestNextAgentHealthWait(t *testing.T) {
	for _, tc := range []struct {
		name     string
		wait     time.Duration
		timedOut bool
		want     time.Duration
	}{
		{"first check timed out", agentHealthFirstWait, true, 2 * agentHealthFirstWait},
		{"a later check timed out", 4 * time.Second, true, 8 * time.Second},
		{"capped", agentHealthMaxWait, true, agentHealthMaxWait},
		{"doubling past the cap stops at it", agentHealthMaxWait/2 + time.Second, true, agentHealthMaxWait},
		{"failed fast", agentHealthFirstWait, false, agentHealthFirstWait},
		{"answered", 2 * time.Second, false, 2 * time.Second},
	} {
		require.Equal(t, tc.want, nextAgentHealthWait(tc.wait, tc.timedOut), tc.name)
	}
}

// A per-test mock whose recorder stated its lifetime (MetadataLifetimeKey) is
// the data of the test it was recorded in: the window filter serves it in its
// own window, and in the startup band when it predates the first test, and
// lax mode never promotes it to the reusable pool, where every other test
// would be served it. A per-test mock recorded before lifetimes were stated is
// promoted as it always was. KEPLOY_STRICT_MOCK_WINDOW=0, the escape hatch for
// recordings that predate classification, leaves a stated mock as stated.
func TestFilterByTimeStampTierAware_StatedPerTestIsNeverPromoted(t *testing.T) {
	prevOverride, prevExplicitOff := strictWindowEnvOverride, strictWindowEnvExplicitOff
	t.Cleanup(func() {
		strictWindowEnvOverride, strictWindowEnvExplicitOff = prevOverride, prevExplicitOff
	})

	first := time.Date(2026, 10, 15, 9, 0, 0, 0, time.UTC) // the first test window starts
	after := first.Add(10 * time.Second)                   // this test's window
	before := after.Add(2 * time.Second)
	mock := func(name string, req time.Time, stated string, lifetime models.Lifetime) *models.Mock {
		m := &models.Mock{
			Version: "api.keploy.io/v1beta1",
			Name:    name,
			Kind:    models.HTTP,
			Spec: models.MockSpec{
				Metadata:         map[string]string{"type": "HTTP_CLIENT"},
				ReqTimestampMock: req,
				ResTimestampMock: req.Add(time.Millisecond),
			},
			TestModeInfo: models.TestModeInfo{Lifetime: lifetime, LifetimeDerived: true},
		}
		if stated != "" {
			m.Spec.Metadata[models.MetadataLifetimeKey] = stated
		}
		return m
	}
	mocks := []*models.Mock{
		mock("stated-startup", first.Add(-time.Second), models.MetadataLifetimePerTest, models.LifetimePerTest),
		mock("stated-earlier-test", first.Add(time.Second), models.MetadataLifetimePerTest, models.LifetimePerTest),
		mock("stated-own", after.Add(time.Second), models.MetadataLifetimePerTest, models.LifetimePerTest),
		mock("stated-later-test", before.Add(time.Second), models.MetadataLifetimePerTest, models.LifetimePerTest),
		mock("stated-session", first.Add(time.Second), models.MetadataLifetimeSession, models.LifetimeSession),
		mock("unstated-earlier-test", first.Add(time.Second), "", models.LifetimePerTest),
	}

	for _, tc := range []struct {
		name        string
		explicitOff bool
		unfiltered  []string
	}{
		{"lax", false, []string{"stated-session", "unstated-earlier-test"}},
		{"KEPLOY_STRICT_MOCK_WINDOW=0", true, []string{"stated-session", "unstated-earlier-test"}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			strictWindowEnvOverride, strictWindowEnvExplicitOff = false, tc.explicitOff
			filtered, unfiltered := filterByTimeStampTierAware(context.Background(), zap.NewNop(), mocks, after, before, false, first)
			assert.ElementsMatch(t, []string{"stated-startup", "stated-own"}, mockNames(filtered),
				"a stated per-test mock is served in its own window and the startup band only")
			assert.ElementsMatch(t, tc.unfiltered, mockNames(unfiltered),
				"lax promotion reaches unstated per-test mocks only")
		})
	}
}

// portServer is an HTTP server on 127.0.0.1 that answers each request with
// its own port, and that port.
func portServer(t *testing.T) (*httptest.Server, uint16) {
	t.Helper()
	var port uint16
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_, _ = fmt.Fprint(w, port)
	}))
	t.Cleanup(srv.Close)
	port = uint16(srv.Listener.Addr().(*net.TCPAddr).Port)
	return srv, port
}

// portTest is a GET of url recorded on the app's port appPort.
func portTest(url string, appPort uint16) *models.TestCase {
	return &models.TestCase{Name: "test-1", Kind: models.HTTP, AppPort: appPort,
		HTTPReq: models.HTTPReq{Method: "GET", URL: url, Header: map[string]string{}}}
}

// A test goes to the port it was recorded on (app_port), over its URL's
// port: an app serving on two ports (HTTP on one, SSE on another) gets each
// test on its own (#3680). Nothing here publishes anything, as on a native
// replay: no PublishedPort.
func TestATestGoesToTheAppPortItWasRecordedOn(t *testing.T) {
	a, pa := portServer(t)
	_, pb := portServer(t)
	for _, tc := range []struct {
		url     string
		appPort uint16
		want    uint16
	}{
		{a.URL + "/x", pb, pb},
		{a.URL + "/x", pa, pa},
		{fmt.Sprintf("http://127.0.0.1:%d/x", pb), pa, pa},
		{a.URL + "/x", 0, pa}, // no app_port: the URL's
	} {
		resp, err := SimulateHTTP(context.Background(), portTest(tc.url, tc.appPort), "test-set-0", zap.NewNop(), SimulationConfig{APITimeout: 5})
		require.NoError(t, err)
		require.Equal(t, fmt.Sprint(tc.want), resp.Body, "url %s, app_port %d", tc.url, tc.appPort)
	}
}

// refusedPort is a port on 127.0.0.1 nothing listens on, held for the test's
// life so that no other socket can take it (testnet.RefusedAddr).
func refusedPort(t *testing.T) uint16 {
	t.Helper()
	_, p, err := net.SplitHostPort(testnet.RefusedAddr(t, "tcp"))
	require.NoError(t, err)
	port, err := strconv.ParseUint(p, 10, 16)
	require.NoError(t, err)
	return uint16(port)
}

// unreachableAt says every address is one the app can never be reached at,
// so a refused test fails at once, saying so.
type unreachableAt struct{}

func (unreachableAt) UnreachableAppPort(context.Context, string, uint16, uint16) string {
	return "it is published on host port 18080 instead"
}

// A test recorded on an app port that the host reaches only through one other
// host port goes to that host port, which PublishedPort names; without it, it
// goes to its app port and fails, saying why. A port the user set
// (ConfigPort) is where it goes, and PublishedPort is not asked.
func TestATestOfAnAppPortPublishedOnOneHostPortGoesThere(t *testing.T) {
	srv, hostPort := portServer(t)
	appPort := refusedPort(t)
	tc := func() *models.TestCase { return portTest(srv.URL+"/x", appPort) }

	_, err := SimulateHTTP(context.Background(), tc(), "test-set-0", zap.NewNop(), SimulationConfig{APITimeout: 5, AppPortReachability: unreachableAt{}})
	require.True(t, IsUnreachableAppPort(err), "without PublishedPort the test goes to its app port: %v", err)
	require.Contains(t, err.Error(), "it is published on host port 18080 instead")

	var asked []string
	published := func(_ context.Context, host string, port uint16) uint16 {
		asked = append(asked, fmt.Sprintf("%s %d", host, port))
		if port == appPort {
			return hostPort
		}
		return 0
	}
	resp, err := SimulateHTTP(context.Background(), tc(), "test-set-0", zap.NewNop(), SimulationConfig{APITimeout: 5, AppPortReachability: unreachableAt{}, PublishedPort: published})
	require.NoError(t, err)
	require.Equal(t, fmt.Sprint(hostPort), resp.Body)
	require.Equal(t, []string{fmt.Sprintf("127.0.0.1 %d", appPort)}, asked)

	asked = nil
	resp, err = SimulateHTTP(context.Background(), tc(), "test-set-0", zap.NewNop(), SimulationConfig{APITimeout: 5, ConfigPort: uint32(hostPort), PublishedPort: published})
	require.NoError(t, err)
	require.Equal(t, fmt.Sprint(hostPort), resp.Body)
	require.Empty(t, asked, "a port the user set is not redirected")

	// A replaceWith that matches the recorded URL and names the port is the
	// user's choice too: asked of the recorded URL, which the resolved one
	// no longer contains.
	recorded := strings.TrimPrefix(srv.URL, "http://")
	_, err = SimulateHTTP(context.Background(), tc(), "test-set-0", zap.NewNop(), SimulationConfig{APITimeout: 5, AppPortReachability: unreachableAt{},
		URLReplacements: map[string]string{recorded: fmt.Sprintf("localhost:%d", appPort)}, PublishedPort: published})
	require.True(t, IsUnreachableAppPort(err), "a matching replaceWith naming the port sent the test to the published port: %v", err)
}

// PublishedTarget moves a target to the host port PublishedPort names only
// when it is the test's app port and nothing the user set moved this test: a
// replaceWith URL or port map that does not apply to it does not stop it.
func TestPublishedTargetMovesOnlyAnUnmovedAppPort(t *testing.T) {
	to18080 := func(_ context.Context, _ string, appPort uint16) uint16 {
		if appPort == 8080 {
			return 18080
		}
		return 0
	}
	none := func(context.Context, string, uint16) uint16 { return 0 }
	for _, tc := range []struct {
		name     string
		cfg      SimulationConfig
		original string // the recorded URL or authority; "" means target
		target   string
		appPort  uint16
		isHTTP   bool
		want     string
	}{
		{"http, published elsewhere", SimulationConfig{PublishedPort: to18080}, "", "http://localhost:8080/x?q=1", 8080, true, "http://localhost:18080/x?q=1"},
		{"http, IPv6 loopback", SimulationConfig{PublishedPort: to18080}, "", "http://[::1]:8080/x", 8080, true, "http://[::1]:18080/x"},
		{"grpc authority", SimulationConfig{PublishedPort: to18080}, "", "localhost:8080", 8080, false, "localhost:18080"},
		{"published as itself, several, or none", SimulationConfig{PublishedPort: none}, "", "http://localhost:8080/x", 8080, true, "http://localhost:8080/x"},
		{"nothing to ask (native)", SimulationConfig{}, "", "http://localhost:8080/x", 8080, true, "http://localhost:8080/x"},
		{"no app port", SimulationConfig{PublishedPort: to18080}, "", "http://localhost:8080/x", 0, true, "http://localhost:8080/x"},
		{"target is not the app port", SimulationConfig{PublishedPort: to18080}, "", "http://localhost:9090/x", 8080, true, "http://localhost:9090/x"},
		{"test.port", SimulationConfig{PublishedPort: to18080, ConfigPort: 8080}, "", "http://localhost:8080/x", 8080, true, "http://localhost:8080/x"},
		{"a replaceWith URL that matches the test and names a port", SimulationConfig{PublishedPort: to18080, URLReplacements: map[string]string{"localhost:8080": "127.0.0.1:8080"}}, "http://localhost:8080/x", "http://127.0.0.1:8080/x", 8080, true, "http://127.0.0.1:8080/x"},
		{"one with its scheme", SimulationConfig{PublishedPort: to18080, URLReplacements: map[string]string{"http://localhost:8080": "http://127.0.0.1:8080"}}, "http://localhost:8080/x", "http://127.0.0.1:8080/x", 8080, true, "http://127.0.0.1:8080/x"},
		{"a replaceWith URL that matches the test and changes only its path", SimulationConfig{PublishedPort: to18080, URLReplacements: map[string]string{"/x": "/y"}}, "http://localhost:8080/x", "http://localhost:8080/y", 8080, true, "http://localhost:18080/y"},
		{"a replaceWith URL that does not match the test", SimulationConfig{PublishedPort: to18080, URLReplacements: map[string]string{"staging.internal:9": "localhost:9"}}, "http://localhost:8080/x", "http://localhost:8080/x", 8080, true, "http://localhost:18080/x"},
		{"a gRPC replaceWith that matches the authority", SimulationConfig{PublishedPort: to18080, URLReplacements: map[string]string{"api:8080": "localhost:8080"}}, "api:8080", "localhost:8080", 8080, false, "localhost:8080"},
		{"a port map of the app port", SimulationConfig{PublishedPort: to18080, PortMappings: map[uint32]uint32{8080: 8080}}, "", "http://localhost:8080/x", 8080, true, "http://localhost:8080/x"},
		{"a port map of another port", SimulationConfig{PublishedPort: to18080, PortMappings: map[uint32]uint32{9090: 9191}}, "", "http://localhost:8080/x", 8080, true, "http://localhost:18080/x"},
		{"the URL names the app", SimulationConfig{PublishedPort: to18080, TargetFromURL: true}, "", "http://localhost:8080/x", 8080, true, "http://localhost:8080/x"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			original := tc.original
			if original == "" {
				original = tc.target
			}
			require.Equal(t, tc.want, PublishedTarget(context.Background(), tc.cfg, original, tc.target, tc.appPort, tc.isHTTP))
		})
	}
}

// Whether a replaceWith value names a port, as both resolvers ask it: after
// the scheme, for HTTP; a path after the host names none.
func TestReplacementNamesPort(t *testing.T) {
	for _, tc := range []struct {
		replacement string
		isHTTP      bool
		want        bool
	}{
		{"app.internal:9090", true, true},
		{"http://app.internal:9090", true, true},
		{"https://app.internal:8443", true, true},
		{"https://app.internal", true, false},
		{"http://app.internal", true, false},
		{"[::1]:8080", false, true},
		{"[::1]", false, false},
		{"app.internal:8080/path", true, false},
		{"app.internal", false, false},
	} {
		if got := replacementNamesPort(tc.replacement, tc.isHTTP); got != tc.want {
			t.Errorf("replacementNamesPort(%q, %v) = %v, want %v", tc.replacement, tc.isHTTP, got, tc.want)
		}
	}
}

// When several replaceWith URL entries match a test, one is applied, the
// most specific, on every run, and PublishedTarget decides by that same one:
// it leaves the test where that entry sends it when the entry names a port,
// and moves it when the entry names none.
func TestSeveralMatchingReplacementsApplyTheMostSpecific(t *testing.T) {
	to18080 := func(_ context.Context, _ string, appPort uint16) uint16 {
		if appPort == 8080 {
			return 18080
		}
		return 0
	}
	const recorded = "http://localhost:8080/x"
	for _, tc := range []struct {
		name      string
		urls      map[string]string
		resolved  string
		published string
	}{
		{"the most specific names a port", map[string]string{"localhost": "app.internal", "localhost:8080": "127.0.0.1:8080"},
			"http://127.0.0.1:8080/x", "http://127.0.0.1:8080/x"},
		{"the most specific names none", map[string]string{"localhost": "127.0.0.1:9090", "localhost:8080/x": "localhost:8080/y"},
			"http://localhost:8080/y", "http://localhost:18080/y"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			for i := 0; i < 64; i++ { // a map's order differs from range to range
				got, err := ResolveTestTarget(recorded, tc.urls, nil, "", 8080, 0, true, zap.NewNop())
				require.NoError(t, err)
				require.Equal(t, tc.resolved, got)
				require.Equal(t, tc.published, PublishedTarget(context.Background(), SimulationConfig{PublishedPort: to18080, URLReplacements: tc.urls}, recorded, got, 8080, true))
			}
		})
	}
}

// A replaceWith value with its scheme that names a port keeps that port: the
// scheme is not taken for a host, and the test's app port does not override
// it.
func TestAReplacementWithItsSchemeKeepsItsPort(t *testing.T) {
	got, err := ResolveTestTarget("http://localhost:8080/x", map[string]string{"http://localhost:8080": "http://127.0.0.1:9090"}, nil, "", 8080, 0, true, zap.NewNop())
	require.NoError(t, err)
	require.Equal(t, "http://127.0.0.1:9090/x", got)
}