package k8s

import (
	"bytes"
	"context"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	kubefake "k8s.io/client-go/kubernetes/fake"
)

func TestPostResource(t *testing.T) {
	// Create a test client
	client := &Client{}

	// Create a fake clientset
	fakeClientset := kubefake.NewSimpleClientset()
	client.SetClientset(fakeClientset)

	// Mock the ExecInPod method
	mockExecCalled := false
	mockExecFunc := func(_ context.Context, namespace, name string, command []string, _ string, _ time.Duration) (*unstructured.Unstructured, error) {
		// Mark that the function was called
		mockExecCalled = true

		// Verify parameters
		assert.Equal(t, "test-pod", name)
		assert.Equal(t, "default", namespace)

		return &unstructured.Unstructured{
			Object: map[string]interface{}{
				"apiVersion": "v1",
				"kind":       "Pod",
				"metadata": map[string]interface{}{
					"name":      name,
					"namespace": namespace,
				},
				"spec": map[string]interface{}{
					"command": command,
				},
				"status": map[string]interface{}{
					"stdout": "test command output",
					"stderr": "",
					"error":  "",
				},
			},
		}, nil
	}

	// Set the mock implementation
	client.SetExecInPodFunc(mockExecFunc)

	// Test cases
	testCases := []struct {
		name         string
		namespace    string
		resourceName string
		subresource  string
		body         map[string]interface{}
		expectError  bool
		errorMsg     string
		gvr          schema.GroupVersionResource
		checkExec    bool
	}{
		{
			name:         "Post to pod exec",
			namespace:    "default",
			resourceName: "test-pod",
			subresource:  "exec",
			body: map[string]interface{}{
				"command": []interface{}{"ls", "-la"},
			},
			expectError: false,
			gvr:         schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
			checkExec:   true,
		},
		{
			name:         "Post to pod exec with string command",
			namespace:    "default",
			resourceName: "test-pod",
			subresource:  "exec",
			body: map[string]interface{}{
				"command": "ls -la",
			},
			expectError: false,
			gvr:         schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
			checkExec:   true,
		},
		{
			name:         "Post to pod exec with container",
			namespace:    "default",
			resourceName: "test-pod",
			subresource:  "exec",
			body: map[string]interface{}{
				"command":   []interface{}{"ls", "-la"},
				"container": "sidecar",
			},
			expectError: false,
			gvr:         schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
			checkExec:   true,
		},
		{
			name:         "Post to pod exec with timeout",
			namespace:    "default",
			resourceName: "test-pod",
			subresource:  "exec",
			body: map[string]interface{}{
				"command": []interface{}{"ls", "-la"},
				"timeout": 30,
			},
			expectError: false,
			gvr:         schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
			checkExec:   true,
		},
		{
			name:         "Post to pod exec missing command",
			namespace:    "default",
			resourceName: "test-pod",
			subresource:  "exec",
			body:         map[string]interface{}{},
			expectError:  true,
			errorMsg:     "command is required for pod exec",
			gvr:          schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
		},
		{
			name:         "Empty resource name",
			namespace:    "default",
			resourceName: "",
			subresource:  "exec",
			body: map[string]interface{}{
				"command": []interface{}{"ls", "-la"},
			},
			expectError: true,
			errorMsg:    "resource name cannot be empty",
			gvr:         schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
		},
	}

	// Run test cases
	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			// Reset the mock flag
			mockExecCalled = false

			// Call the method
			ctx := context.Background()
			result, err := client.PostResource(ctx, tc.gvr, tc.namespace, tc.resourceName, tc.subresource, tc.body, nil)

			// Assert expectations
			if tc.expectError {
				assert.Error(t, err)
				if tc.errorMsg != "" {
					assert.Contains(t, err.Error(), tc.errorMsg)
				}
			} else {
				assert.NoError(t, err)
				assert.NotNil(t, result)

				// Verify the resource name
				assert.Equal(t, tc.resourceName, result.GetName())

				// Verify the namespace if applicable
				if tc.namespace != "" {
					assert.Equal(t, tc.namespace, result.GetNamespace())
				}

				// Check if the mock was called for exec tests
				if tc.checkExec {
					assert.True(t, mockExecCalled, "ExecInPod function should have been called")

					// Check for exec output
					stdout, found, err := unstructured.NestedString(result.Object, "status", "stdout")
					assert.NoError(t, err)
					assert.True(t, found)
					assert.Equal(t, "test command output", stdout)
				}
			}
		})
	}
}

func TestExecInPod(t *testing.T) {
	// Create a test client
	client := &Client{}

	// Create a fake clientset
	fakeClientset := kubefake.NewSimpleClientset()
	client.SetClientset(fakeClientset)

	// Mock the ExecInPod method
	var capturedTimeout time.Duration
	mockExecFunc := func(_ context.Context, namespace, name string, command []string, _ string, timeout time.Duration) (*unstructured.Unstructured, error) {
		// Apply the same timeout adjustments as the real implementation
		if timeout <= 0 {
			timeout = 15 * time.Second
		}
		if timeout > MaxExecTimeout {
			timeout = MaxExecTimeout
		}

		// Capture the adjusted timeout for verification
		capturedTimeout = timeout

		// Perform the same validation as the real implementation
		if name == "" {
			return nil, fmt.Errorf("pod name cannot be empty")
		}

		if len(command) == 0 {
			return nil, fmt.Errorf("command cannot be empty")
		}

		// Return a mock result
		return &unstructured.Unstructured{
			Object: map[string]interface{}{
				"apiVersion": "v1",
				"kind":       "Pod",
				"metadata": map[string]interface{}{
					"name":      name,
					"namespace": namespace,
				},
				"spec": map[string]interface{}{
					"command": command,
				},
				"status": map[string]interface{}{
					"stdout": "test command output",
					"stderr": "",
					"error":  "",
				},
			},
		}, nil
	}

	// Set the mock implementation
	client.SetExecInPodFunc(mockExecFunc)

	// Test cases
	testCases := []struct {
		name            string
		namespace       string
		podName         string
		command         []string
		container       string
		timeout         time.Duration
		expectError     bool
		errorMsg        string
		checkTimeout    bool
		expectedTimeout time.Duration
	}{
		{
			name:        "Empty pod name",
			namespace:   "default",
			podName:     "",
			command:     []string{"ls", "-la"},
			timeout:     30 * time.Second,
			expectError: true,
			errorMsg:    "pod name cannot be empty",
		},
		{
			name:        "Empty command",
			namespace:   "default",
			podName:     "test-pod",
			command:     []string{},
			timeout:     30 * time.Second,
			expectError: true,
			errorMsg:    "command cannot be empty",
		},
		{
			name:            "Zero timeout",
			namespace:       "default",
			podName:         "test-pod",
			command:         []string{"ls", "-la"},
			timeout:         0,
			expectError:     false,
			checkTimeout:    true,
			expectedTimeout: 15 * time.Second, // Default timeout
		},
		{
			name:            "Negative timeout",
			namespace:       "default",
			podName:         "test-pod",
			command:         []string{"ls", "-la"},
			timeout:         -5 * time.Second,
			expectError:     false,
			checkTimeout:    true,
			expectedTimeout: 15 * time.Second, // Default timeout
		},
		{
			name:            "Excessive timeout",
			namespace:       "default",
			podName:         "test-pod",
			command:         []string{"ls", "-la"},
			timeout:         10 * time.Minute,
			expectError:     false,
			checkTimeout:    true,
			expectedTimeout: MaxExecTimeout, // Capped at MaxExecTimeout
		},
		{
			name:        "With container",
			namespace:   "default",
			podName:     "test-pod",
			command:     []string{"ls", "-la"},
			container:   "sidecar",
			timeout:     30 * time.Second,
			expectError: false,
		},
	}

	// Run test cases
	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			// Reset the captured timeout
			capturedTimeout = 0

			// Call the method
			ctx := context.Background()
			result, err := client.ExecInPod(ctx, tc.namespace, tc.podName, tc.command, tc.container, tc.timeout)

			// Assert expectations
			if tc.expectError {
				assert.Error(t, err)
				if tc.errorMsg != "" {
					assert.Contains(t, err.Error(), tc.errorMsg)
				}
			} else {
				assert.NoError(t, err)
				assert.NotNil(t, result)

				// Check timeout handling if applicable
				if tc.checkTimeout {
					assert.Equal(t, tc.expectedTimeout, capturedTimeout, "Timeout should be adjusted correctly")
				}
			}
		})
	}
}

// Regression tests for the boundedBuffer used to cap pod-exec output size.
// See GHSA-qw5r-ppcg-f8rj for the analogous pod-log fix.

func TestBoundedBuffer_WriteUnderLimit(t *testing.T) {
	t.Parallel()

	b := newBoundedBuffer(1024)
	n, err := b.Write([]byte("hello"))
	require.NoError(t, err)
	assert.Equal(t, 5, n)
	assert.Equal(t, "hello", b.String())
	assert.False(t, b.Truncated())
}

func TestBoundedBuffer_WriteAtExactLimit(t *testing.T) {
	t.Parallel()

	payload := bytes.Repeat([]byte("a"), 1024)
	b := newBoundedBuffer(1024)
	n, err := b.Write(payload)
	require.NoError(t, err)
	assert.Equal(t, 1024, n)
	assert.Equal(t, 1024, len(b.String()))
	assert.False(t, b.Truncated())
}

func TestBoundedBuffer_WriteOverLimit_TruncatesAndMarks(t *testing.T) {
	t.Parallel()

	payload := bytes.Repeat([]byte("a"), 4096)
	b := newBoundedBuffer(1024)
	n, err := b.Write(payload)
	require.NoError(t, err)
	// Caller sees the full write to keep the stream upstream happy.
	assert.Equal(t, 4096, n)
	// Internally we only buffered up to the cap.
	assert.Equal(t, 1024, len(b.String()))
	assert.True(t, b.Truncated())
}

func TestBoundedBuffer_MultipleWritesAccumulate(t *testing.T) {
	t.Parallel()

	b := newBoundedBuffer(10)
	_, err := b.Write([]byte("hello"))
	require.NoError(t, err)
	assert.False(t, b.Truncated())

	// Second write fits exactly.
	_, err = b.Write([]byte("world"))
	require.NoError(t, err)
	assert.Equal(t, "helloworld", b.String())
	assert.False(t, b.Truncated())

	// Third write is entirely beyond the cap.
	_, err = b.Write([]byte("!"))
	require.NoError(t, err)
	assert.Equal(t, "helloworld", b.String(), "no further bytes should be buffered")
	assert.True(t, b.Truncated())
}

func TestBoundedBuffer_PartialOverflow(t *testing.T) {
	t.Parallel()

	b := newBoundedBuffer(10)
	// First write fills 5 of 10.
	_, err := b.Write([]byte("hello"))
	require.NoError(t, err)
	// Second write would put us 5 over: only the first 5 should be buffered.
	_, err = b.Write([]byte("WORLD!!!!!"))
	require.NoError(t, err)
	assert.Equal(t, "helloWORLD", b.String())
	assert.True(t, b.Truncated())
}

func TestBoundedBuffer_AsIOWriter(t *testing.T) {
	t.Parallel()

	// A boundedBuffer must satisfy io.Writer for use with
	// remotecommand.StreamOptions. Use strings.NewReader to mimic the
	// streamed-write pattern.
	b := newBoundedBuffer(100)
	r := strings.NewReader(strings.Repeat("x", 250))
	buf := make([]byte, 32)
	for {
		n, err := r.Read(buf)
		if n > 0 {
			_, werr := b.Write(buf[:n])
			require.NoError(t, werr)
		}
		if err != nil {
			break
		}
	}
	assert.Equal(t, 100, len(b.String()))
	assert.True(t, b.Truncated())
}
