// +build integration

/*
 * 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"
	"os"
	"path/filepath"
	"testing"

	"github.com/pingcap/mimir/model"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"github.com/pingcap/mimir/pkg/protos"
	"github.com/stretchr/testify/require"
	corev1 "k8s.io/api/core/v1"
	k8s "k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

const (
	simpleStreamingSql = `CREATE TABLE datagen (
			f0 INT,
			f1 INT,
			f2 STRING
		) WITH (
			'connector' = 'datagen',
			'rows-per-second'= '1'
		);
		CREATE TABLE print1 (
			f0 INT,
			f1 INT,
			f2 STRING
		) WITH (
			'connector' = 'print'
		);
		INSERT INTO print1 SELECT * FROM datagen;
		`
)

func buildTestAPIService() (*apiService, *k8s.Clientset, error) {
	kubeconfig := filepath.Join(homedir.HomeDir(), ".kube", "config")
	// use the current context in kubeconfig
	config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
	if err != nil {
		return nil, nil, err
	}
	clientSet, err := k8s.NewForConfig(config)
	if err != nil {
		return nil, nil, err
	}
	flinkClientPath := os.Getenv("FLINK_CLIENT_PATH")
	flinkImage := os.Getenv("FLINK_IMAGE")
	db, err := model.Dial(&model.Config{
		Driver: "mysql",
		DSN:    os.Getenv("TIDB_DSN"),
	})
	if err != nil {
		return nil, nil, err
	}
	dao, err := model.NewJobDao(model.WithDB(db))
	svc, err := New(
		K8sClientOpt(clientSet),
		FlinkClientPathOpt(flinkClientPath),
		FlinkImageOpt(flinkImage),
		JobDaoOpt(dao))
	if err != nil {
		return nil, nil, err
	}
	return svc, clientSet, nil
}

func TestApiService_runApplication(t *testing.T) {
	assert := require.New(t)
	svc, _, err := buildTestAPIService()
	assert.Nil(err)
	id, _, err := svc.runApplication(context.Background(), "", "foo", &corev1.Secret{})
	assert.Nil(err)
	assert.Equal("foo", id)
	_, _, err = svc.runApplication(context.Background(), "", "foo", &corev1.Secret{})
	assert.Nil(err)
}

func TestApiService_SubmitJob(t *testing.T) {
	assert := require.New(t)
	svc, _, err := buildTestAPIService()
	assert.Nil(err)
	param := protos.SubmitJobParam{
		SQL: simpleStreamingSql,
	}
	data, err := svc.SubmitJob(context.Background(), &param)
	assert.Nil(err)
	assert.Greater(0, len(data.ClusterID))
}

func TestApiService_ListJob(t *testing.T) {
	assert := require.New(t)
	svc, _, err := buildTestAPIService()
	assert.Nil(err)
	data, err := svc.ListJob(context.Background(), &protos.ListJobParam{})
	assert.Nil(err)
	assert.Greater(0, len(data.Items))
}

func TestApiService_setOwnerReference2(t *testing.T) {
	assert := require.New(t)
	svc, cli, err := buildTestAPIService()
	assert.Nil(err)
	ctx := context.Background()
	sct, err := cli.CoreV1().Secrets("default").Get(ctx, "mimir-test", metav1.GetOptions{})
	assert.Nil(err)
	err = svc.setOwnerReference(ctx, "default", sct, "foo")
	assert.Nil(err)
}

func TestApiService_queryFlinkJobs(t *testing.T) {
	assert := require.New(t)

	svc, err := New(FlinkClientPathOpt(os.Getenv("FLINK_CLIENT_PATH")))
	assert.Nil(err)
	assert.NotNil(svc)

	// TODO: job should be submitted first
	flinkJobIds, err := svc.queryFlinkJobs(context.Background(), "default", "cluster0")
	assert.Nil(err)
	assert.NotEmpty(flinkJobIds)
	t.Log(flinkJobIds)
}