// Copyright 2024 Woodpecker Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package kubernetes

import (
	"context"
	"fmt"
	"runtime"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/urfave/cli/v3"
	kube_core_v1 "k8s.io/api/core/v1"
	kube_meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes/fake"

	"go.woodpecker-ci.org/woodpecker/v3/pipeline/backend/types"
)

func TestGettingConfig(t *testing.T) {
	engine := kube{
		config: &config{
			Namespace:            "default",
			StorageClass:         "hdd",
			VolumeSize:           "1G",
			StorageRwx:           false,
			PodLabels:            map[string]string{"l1": "v1"},
			PodAnnotations:       map[string]string{"a1": "v1"},
			ImagePullSecretNames: []string{"regcred"},
			SecurityContext:      SecurityContextConfig{RunAsNonRoot: false},
		},
	}
	config := engine.getConfig()
	config.Namespace = "wp"
	config.StorageClass = "ssd"
	config.StorageRwx = true
	config.PodLabels = nil
	config.PodAnnotations["a2"] = "v2"
	config.ImagePullSecretNames = append(config.ImagePullSecretNames, "docker.io")
	config.SecurityContext.RunAsNonRoot = true

	assert.Equal(t, "default", engine.config.Namespace)
	assert.Equal(t, "hdd", engine.config.StorageClass)
	assert.Equal(t, "1G", engine.config.VolumeSize)
	assert.False(t, engine.config.StorageRwx)
	assert.Len(t, engine.config.PodLabels, 1)
	assert.Len(t, engine.config.PodAnnotations, 1)
	assert.Len(t, engine.config.ImagePullSecretNames, 1)
	assert.False(t, engine.config.SecurityContext.RunAsNonRoot)
}

func TestSetupWorkflow(t *testing.T) {
	namespace := "foo"
	volumeName := "volume-name"
	volumePath := volumeName + ":/woodpecker"
	networkName := "test-network"
	taskUUID := "11301"

	engine := kube{
		config: &config{
			Namespace:            namespace,
			StorageClass:         "hdd",
			VolumeSize:           "1G",
			StorageRwx:           false,
			PodLabels:            map[string]string{"l1": "v1"},
			PodAnnotations:       map[string]string{"a1": "v1"},
			ImagePullSecretNames: []string{"regcred"},
			SecurityContext:      SecurityContextConfig{RunAsNonRoot: false},
		},
		client: fake.NewClientset(),
	}

	serviceWithPorts := types.Step{
		OrgID:    42,
		Name:     "service",
		UUID:     "123",
		Type:     types.StepTypeService,
		Volumes:  []string{volumePath},
		Networks: []types.Conn{{Name: networkName, Aliases: []string{"alias"}}},
		Ports: []types.Port{
			{Number: 8080, Protocol: "tcp"},
		},
	}

	conf := &types.Config{
		Volume:  volumePath,
		Network: networkName,
		Stages: []*types.Stage{
			{
				Steps: []*types.Step{
					&serviceWithPorts,
					{
						OrgID:    42,
						UUID:     "234",
						Name:     "service2",
						Type:     types.StepTypeService,
						Volumes:  []string{volumePath},
						Networks: []types.Conn{{Name: networkName, Aliases: []string{"alias"}}},
					},
				},
			},
			{
				Steps: []*types.Step{
					{
						OrgID:    42,
						UUID:     "456",
						Name:     "step-1",
						Volumes:  []string{volumePath},
						Networks: []types.Conn{{Name: networkName, Aliases: []string{"alias"}}},
					},
				},
			},
		},
	}

	err := engine.SetupWorkflow(context.Background(), conf, taskUUID)
	assert.NoError(t, err, "SetupWorkflow should not error with minimal config and fake client")

	_, err = engine.client.CoreV1().PersistentVolumeClaims(namespace).Get(context.Background(), "volume-name", kube_meta_v1.GetOptions{})
	assert.NoError(t, err, "persistent volume should be created during workflow setup")

	_, err = engine.client.CoreV1().Services(namespace).Get(context.Background(), "wp-hsvc-"+taskUUID, kube_meta_v1.GetOptions{})
	assert.NoError(t, err, "headless service should be created during workflow setup")
}

func TestAffinityFromCliContext(t *testing.T) {
	t.Setenv("WOODPECKER_BACKEND_K8S_NAMESPACE", "")
	t.Setenv("WOODPECKER_BACKEND_K8S_POD_AFFINITY", `{
		"podAffinity": {
			"requiredDuringSchedulingIgnoredDuringExecution": [
			{
				"labelSelector": {},
				"matchLabelKeys": [
				"woodpecker-ci.org/task-uuid"
				],
				"topologyKey": "kubernetes.io/hostname"
			}
			]
		}
		}`)
	t.Setenv("WOODPECKER_BACKEND_K8S_POD_AFFINITY_ALLOW_FROM_STEP", "false")

	cmd := &cli.Command{
		Flags: Flags,
		Action: func(ctx context.Context, c *cli.Command) error {
			ctx = context.WithValue(ctx, types.CliCommand, c)
			config, err := configFromCliContext(ctx)

			require.NoError(t, err)
			require.NotNil(t, config)
			assert.False(t, config.PodAffinityAllowFromStep)

			// Verify affinity was parsed
			require.NotNil(t, config.PodAffinity)
			require.NotNil(t, config.PodAffinity.PodAffinity)
			require.Len(t, config.PodAffinity.PodAffinity.RequiredDuringSchedulingIgnoredDuringExecution, 1)

			term := config.PodAffinity.PodAffinity.RequiredDuringSchedulingIgnoredDuringExecution[0]
			assert.Equal(t, "kubernetes.io/hostname", term.TopologyKey)
			assert.Equal(t, []string{"woodpecker-ci.org/task-uuid"}, term.MatchLabelKeys)

			return nil
		},
	}
	err := cmd.Run(context.Background(), []string{"test"})
	require.NoError(t, err)
}

func makeStep(uuid string) *types.Step {
	return &types.Step{
		UUID:  uuid,
		Name:  "step-" + uuid,
		OrgID: 1,
	}
}

func makeEngine(client *fake.Clientset) *kube {
	return &kube{
		client: client,
		config: &config{
			Namespace: "test-ns",
		},
	}
}

func createPod(
	t *testing.T,
	client *fake.Clientset,
	step *types.Step,
	namespace string,
) string {
	t.Helper()
	podName, err := stepToPodName(step)
	require.NoError(t, err)

	pod := &kube_core_v1.Pod{
		ObjectMeta: kube_meta_v1.ObjectMeta{
			Name:      podName,
			Namespace: namespace,
		},
		Status: kube_core_v1.PodStatus{
			Phase: kube_core_v1.PodPending,
		},
	}
	_, err = client.CoreV1().Pods(namespace).Create(
		context.Background(), pod, kube_meta_v1.CreateOptions{},
	)
	require.NoError(t, err)
	return podName
}

func TestWaitStepReturnsOnContextCancel(t *testing.T) {
	client := fake.NewClientset()
	engine := makeEngine(client)
	step := makeStep("ctx-cancel-01")
	namespace := "test-ns"

	createPod(t, client, step, namespace)

	ctx, cancel := context.WithCancelCause(context.Background())

	type result struct {
		state *types.State
		err   error
	}
	ch := make(chan result, 1)

	go func() {
		s, err := engine.WaitStep(ctx, step, "task-1")
		ch <- result{s, err}
	}()

	// Give the informer time to start and begin watching.
	time.Sleep(200 * time.Millisecond)

	cancel(nil)

	select {
	case r := <-ch:
		assert.Nil(t, r.state)
		assert.ErrorIs(t, r.err, context.Canceled)
	case <-time.After(3 * time.Second):
		t.Fatal("WaitStep did not return after context cancellation")
	}
}

func TestWaitStepReturnsOnAlreadyDeletedPod(t *testing.T) {
	client := fake.NewClientset()
	engine := makeEngine(client)
	step := makeStep("pod-delete-02")
	namespace := "test-ns"

	podName := createPod(t, client, step, namespace)

	// Delete before WaitStep starts
	err := client.CoreV1().Pods(namespace).Delete(context.Background(), podName, kube_meta_v1.DeleteOptions{})
	require.NoError(t, err)

	type result struct {
		state *types.State
		err   error
	}
	ch := make(chan result, 1)
	go func() {
		s, err := engine.WaitStep(context.Background(), step, "task-1")
		ch <- result{s, err}
	}()

	select {
	case r := <-ch:
		require.NoError(t, r.err)
		require.NotNil(t, r.state)
		assert.True(t, r.state.Exited)
		assert.Equal(t, 0, r.state.ExitCode)
	case <-time.After(3 * time.Second):
		t.Fatal("WaitStep did not return for already-deleted pod")
	}
}

func TestWaitStepNoGoroutineLeak(t *testing.T) {
	client := fake.NewClientset()
	engine := makeEngine(client)
	namespace := "test-ns"
	numSteps := 10

	steps := make([]*types.Step, numSteps)
	for i := range numSteps {
		steps[i] = makeStep(fmt.Sprintf("leak-%02d", i))
		createPod(t, client, steps[i], namespace)
	}

	runtime.GC()
	time.Sleep(100 * time.Millisecond)
	baselineGoroutines := runtime.NumGoroutine()

	var wg sync.WaitGroup
	for i := range numSteps {
		wg.Add(1)
		go func() {
			defer wg.Done()

			ctx, cancel := context.WithCancelCause(context.Background())

			go func() {
				_, _ = engine.WaitStep(ctx, steps[i], fmt.Sprintf("task-%d", i))
			}()

			time.Sleep(200 * time.Millisecond)
			cancel(nil)
		}()
	}
	wg.Wait()

	time.Sleep(1 * time.Second)

	afterCancelGoroutines := runtime.NumGoroutine()
	leaked := afterCancelGoroutines - baselineGoroutines

	assert.Less(t, leaked, numSteps,
		"goroutines leaked after canceling %d WaitStep calls: got %d leaked",
		numSteps, leaked)
}
