/*
 * Copyright 2021 PingCAP, Inc.
 *
 * 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,
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package service

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes/fake"
	k8stesting "k8s.io/client-go/testing"

	"github.com/pingcap/mimir/model"
	"github.com/pingcap/mimir/pkg/protos"
)

func TestApiService_createSecretDuplicate(t *testing.T) {
	assert := require.New(t)
	fakeClientSet := fake.NewSimpleClientset()
	svc, err := New(
		K8sClientOpt(fakeClientSet),
	)
	assert.Nil(err)
	ctx := context.Background()
	sct1, err := svc.createSecret(ctx, "default", "foo", `SELECT "hello world"`)
	assert.Nil(err)
	assert.Equal("foo", sct1.GetName())
	assert.Equal("default", sct1.GetNamespace())
	assert.Equal(`SELECT "hello world"`, sct1.StringData["input.sql"])
	sct2, err := svc.createSecret(ctx, "default", "foo", `SELECT "hello world"`)
	assert.Nil(err)
	assert.Equal("foo", sct2.GetName())
	assert.Equal("default", sct2.GetNamespace())
	assert.Equal(`SELECT "hello world"`, sct2.StringData["input.sql"])
	actions := fakeClientSet.Actions()
	assert.Greater(len(actions), 0)
	assert.True(actions[0].Matches("create", "secrets"))
	assert.Equal("default", actions[0].GetNamespace())
}

func TestApiService_setOwnerReference(t *testing.T) {
	assert := require.New(t)
	sct := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "default"},
		StringData: map[string]string{"input.sql": `SELECT "hello world"`},
	}
	deployFoo := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "default", UID: "foo"},
	}
	deployBar := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{Name: "bar", Namespace: "default", UID: "bar"},
	}
	fakeClientSet := fake.NewSimpleClientset([]runtime.Object{sct, deployFoo, deployBar}...)
	svc, err := New(
		K8sClientOpt(fakeClientSet),
	)
	assert.Nil(err)
	ctx := context.Background()
	err = svc.setOwnerReference(ctx, "default", sct, "foo")
	assert.Nil(err)
	err = svc.setOwnerReference(ctx, "default", sct, "bar")
	assert.Nil(err)
	actions := fakeClientSet.Actions()
	assert.Equal(len(actions), 4)
	assert.True(actions[1].Matches("patch", "secrets"))
	{
		patchAction, ok := actions[1].(k8stesting.PatchAction)
		assert.True(ok)
		assert.Equal(types.StrategicMergePatchType, patchAction.GetPatchType())
	}
	assert.True(actions[3].Matches("patch", "secrets"))
	{
		patchAction, ok := actions[3].(k8stesting.PatchAction)
		assert.True(ok)
		assert.Equal(types.StrategicMergePatchType, patchAction.GetPatchType())
	}
	assert.Equal("default", actions[1].GetNamespace())
	assert.Equal("default", actions[3].GetNamespace())
	ret, err := fakeClientSet.CoreV1().Secrets("default").Get(ctx, "foo", metav1.GetOptions{})
	assert.Nil(err)
	assert.Equal(2, len(ret.GetOwnerReferences()))
	assert.Equal(types.UID("bar"), ret.GetOwnerReferences()[0].UID)
	assert.Equal(types.UID("foo"), ret.GetOwnerReferences()[1].UID)
}

type ListFooDao struct {
	model.BaseMock
}

func (dao *ListFooDao) List(ctx context.Context, cond *model.ListJobCond) ([]model.Job, error) {
	return []model.Job{
		{
			UserID:     protos.DefaultUserID,
			JobID:      "flink-cluster-fb7aafaccab1530be19ded9932ee731",
			JMEndpoint: "localhost:8081",
			Namespace:  protos.DefaultNamespace,
			JobType:    1,
			InputType:  1,
			InputKey:   "uuid",
			CreatedAt:  time.Now().Local(),
		},
	}, nil
}

func (dao *ListFooDao) Delete(ctx context.Context, userID string, jobID string) error { return nil }

func (dao *ListFooDao) Update(ctx context.Context, job *model.Job) error { return nil }

func TestApiService_MockListJob(t *testing.T) {
	assert := require.New(t)
	serviceflink := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{Name: "flink-cluster-fb7aafaccab1530be19ded9932ee731-rest",
			Namespace: protos.DefaultNamespace,
			Labels:    map[string]string{"type": "flink-native-kubernetes"}},
		Spec: corev1.ServiceSpec{
			Ports: []corev1.ServicePort{
				{
					Name: "rest",
					Port: 8081,
				},
			},
			ClusterIP: "127.0.0.1",
		},
	}

	fakeClientSet := fake.NewSimpleClientset([]runtime.Object{serviceflink}...)
	svc, err := New(
		K8sClientOpt(fakeClientSet),
		JobDaoOpt(&ListFooDao{}),
	)
	assert.Nil(err)
	ctx := context.Background()
	data, err := svc.ListJob(ctx, new(protos.ListJobParam))
	assert.Nil(err)
	assert.NotNil(data)
	assert.Len(data.Items, 1)
	assert.Equal(string("UNKNOWN"), data.Items[0].Status)
	assert.Equal("flink-cluster-fb7aafaccab1530be19ded9932ee731", data.Items[0].JobID)
}