package schedule

import (
	"context"
	"fmt"
	"testing"

	workflowpkg "github.com/argoproj/argo-workflows/v3/pkg/apiclient/workflow"
	wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
	commM "github.com/pingcap/test-infra/common/model"
	"github.com/pingcap/test-infra/services/tps/pkg/common"
	"github.com/stretchr/testify/require"
	"google.golang.org/grpc"
)

var idcArgoURL = "http://argo.idc"
var idcArgoCfg = ArgoConfig{Namespace: "ns-idc", ServiceAccount: "argo-idc", ArtifactRepoRef: &wfapi.ArtifactRepositoryRef{ConfigMap: "argo", Key: "idc"}, URL: idcArgoURL, SetupImage: "hub.idc/qa/setup"}
var idcRmsURL = "http://rms.idc"
var awsArgoURL = "https://argo.aws"
var awsArgoCfg = ArgoConfig{Namespace: "ns-aws", ServiceAccount: "argo-aws", ArtifactRepoRef: &wfapi.ArtifactRepositoryRef{ConfigMap: "argo", Key: "aws"}, URL: awsArgoURL, SetupImage: "hub.aws/qa/setup"}
var awsRmsURL = "https://rms.aws"
var badArgoCfg = ArgoConfig{URL: "http://argo.idc:aaa\\"}

func newArgoByRmsMgr() *ArgoByRms {
	selector := map[string]string{
		idcRmsURL: idcArgoCfg.URL,
		awsRmsURL: awsArgoCfg.URL,
	}
	mgr, err := NewArgoByRms([]ArgoConfig{idcArgoCfg, awsArgoCfg}, selector)
	if err != nil {
		panic(err)
	}
	return mgr
}

func TestNewArgoByRms(t *testing.T) {
	selector := map[string]string{
		idcRmsURL: idcArgoCfg.URL,
		awsRmsURL: awsArgoCfg.URL,
	}
	mgr, err := NewArgoByRms([]ArgoConfig{idcArgoCfg, awsArgoCfg}, selector)
	require.NoError(t, err)
	var poolKeys []string
	for key := range mgr.pools {
		poolKeys = append(poolKeys, key)
	}
	require.ElementsMatch(t, []string{idcArgoCfg.URL, awsArgoCfg.URL}, poolKeys)

	mgr, err = NewArgoByRms([]ArgoConfig{idcArgoCfg, awsArgoCfg, badArgoCfg}, selector)
	require.Error(t, err)
	require.Nil(t, mgr)
}

func TestArgoByRms_ClientFor(t *testing.T) {
	mgr := newArgoByRmsMgr()

	for _, s := range []struct {
		name      string
		rmsUrl    string
		clientTag string
		expected  string
		err       error
	}{
		{
			name:     "aws rms url",
			rmsUrl:   awsRmsURL,
			expected: awsArgoCfg.URL,
		},
		{
			name:      "with idc tag",
			rmsUrl:    awsRmsURL,
			clientTag: idcArgoCfg.URL,
			expected:  idcArgoCfg.URL,
		},
		{
			name:      "with not existed tag",
			rmsUrl:    awsRmsURL,
			clientTag: "not.existed",
			err:       fmt.Errorf("argo client not found for: not.existed"),
		},
	} {
		t.Run(s.name, func(t *testing.T) {
			job := &common.ArgoJob{
				PlanExec: &commM.TestPlanExecution{Meta: map[string]string{ArgoClientKey: s.clientTag}},
				Plan:     &commM.TestPlan{TestPlanBase: commM.TestPlanBase{TestBed: commM.TestBed{RmsUrl: s.rmsUrl}}},
			}
			clientFor, err := mgr.ClientFor(job)
			if s.err != nil {
				require.Equal(t, s.err, err)
			} else {
				require.NoError(t, err)
				require.Equal(t, s.expected, clientFor.URL())
			}
		})
	}
}

func TestArgoByRms_BindClient(t *testing.T) {
	mgr := newArgoByRmsMgr()
	oldRmsUrl := idcRmsURL
	job := &common.ArgoJob{
		PlanExec: &commM.TestPlanExecution{Meta: map[string]string{}},
		Plan:     &commM.TestPlan{TestPlanBase: commM.TestPlanBase{TestBed: commM.TestBed{RmsUrl: oldRmsUrl}}},
	}
	oldClient, err := mgr.ClientFor(job)
	require.NoError(t, err)
	job.Plan.TestBed.RmsUrl = awsRmsURL
	newClient, err := mgr.ClientFor(job)
	require.NoError(t, err)
	require.NotEqual(t, oldClient, newClient)

	job = mgr.BindClient(job, newClient)
	job.Plan.TestBed.RmsUrl = oldRmsUrl
	finalClient, err := mgr.ClientFor(job)
	require.NoError(t, err)
	require.Equal(t, newClient, finalClient)
}

func TestArgoClientFromConfig(t *testing.T) {
	client, err := ArgoClientFromConfig(awsArgoCfg)
	require.NoError(t, err)
	require.Equal(t, awsArgoCfg, client.config)
	require.NotNil(t, client.wfClient)

	_, err = ArgoClientFromConfig(badArgoCfg)
	require.Error(t, err)
}

func TestArgoClient_URL(t *testing.T) {
	client, err := ArgoClientFromConfig(awsArgoCfg)
	require.NoError(t, err)
	require.Equal(t, awsArgoCfg.URL, client.URL())
}

type wfClientStub struct {
	workflowCreateRequest *workflowpkg.WorkflowCreateRequest
	workflowGetRequest    *workflowpkg.WorkflowGetRequest
	workflowStopRequest   *workflowpkg.WorkflowStopRequest
}

func (w *wfClientStub) CreateWorkflow(ctx context.Context, in *workflowpkg.WorkflowCreateRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	w.workflowCreateRequest = in
	return in.Workflow, nil
}

func (w *wfClientStub) GetWorkflow(ctx context.Context, in *workflowpkg.WorkflowGetRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	w.workflowGetRequest = in
	return nil, nil
}

func (w *wfClientStub) ListWorkflows(ctx context.Context, in *workflowpkg.WorkflowListRequest, opts ...grpc.CallOption) (*wfapi.WorkflowList, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) WatchWorkflows(ctx context.Context, in *workflowpkg.WatchWorkflowsRequest, opts ...grpc.CallOption) (workflowpkg.WorkflowService_WatchWorkflowsClient, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) WatchEvents(ctx context.Context, in *workflowpkg.WatchEventsRequest, opts ...grpc.CallOption) (workflowpkg.WorkflowService_WatchEventsClient, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) DeleteWorkflow(ctx context.Context, in *workflowpkg.WorkflowDeleteRequest, opts ...grpc.CallOption) (*workflowpkg.WorkflowDeleteResponse, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) RetryWorkflow(ctx context.Context, in *workflowpkg.WorkflowRetryRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) ResubmitWorkflow(ctx context.Context, in *workflowpkg.WorkflowResubmitRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) ResumeWorkflow(ctx context.Context, in *workflowpkg.WorkflowResumeRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) SuspendWorkflow(ctx context.Context, in *workflowpkg.WorkflowSuspendRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) TerminateWorkflow(ctx context.Context, in *workflowpkg.WorkflowTerminateRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) StopWorkflow(ctx context.Context, in *workflowpkg.WorkflowStopRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	w.workflowStopRequest = in
	return nil, nil
}

func (w *wfClientStub) SetWorkflow(ctx context.Context, in *workflowpkg.WorkflowSetRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) LintWorkflow(ctx context.Context, in *workflowpkg.WorkflowLintRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) PodLogs(ctx context.Context, in *workflowpkg.WorkflowLogRequest, opts ...grpc.CallOption) (workflowpkg.WorkflowService_PodLogsClient, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) WorkflowLogs(ctx context.Context, in *workflowpkg.WorkflowLogRequest, opts ...grpc.CallOption) (workflowpkg.WorkflowService_WorkflowLogsClient, error) {
	//TODO implement me
	panic("implement me")
}

func (w *wfClientStub) SubmitWorkflow(ctx context.Context, in *workflowpkg.WorkflowSubmitRequest, opts ...grpc.CallOption) (*wfapi.Workflow, error) {
	//TODO implement me
	panic("implement me")
}

func TestDecorateWorkflow(t *testing.T) {
	wf := &wfapi.Workflow{}
	wf = decorateWorkflow(wf, idcArgoCfg)
	require.Equal(t, idcArgoCfg.Namespace, wf.Namespace)
	require.Equal(t, idcArgoCfg.ArtifactRepoRef, wf.Spec.ArtifactRepositoryRef)
	require.Equal(t, idcArgoCfg.ServiceAccount, wf.Spec.ServiceAccountName)
	require.Equal(t, 1, len(wf.Spec.Arguments.Parameters))
	arg := wf.Spec.Arguments.Parameters[0]
	require.Equal(t, setupImageParamName, arg.Name)
	require.Equal(t, idcArgoCfg.SetupImage, arg.Value.String())
}

func TestArgoClient_Create(t *testing.T) {
	client, err := ArgoClientFromConfig(awsArgoCfg)
	require.NoError(t, err)
	stub := &wfClientStub{}
	client.wfClient = stub
	wf := &wfapi.Workflow{}
	created, err := client.Create(context.TODO(), wf)
	require.NoError(t, err)
	require.Equal(t, awsArgoCfg.Namespace, stub.workflowCreateRequest.Namespace)
	require.Equal(t, awsArgoCfg.Namespace, created.Namespace)
	require.Equal(t, awsArgoCfg.ArtifactRepoRef, created.Spec.ArtifactRepositoryRef)
	require.Equal(t, awsArgoCfg.ServiceAccount, created.Spec.ServiceAccountName)
}

func TestArgoClient_Get(t *testing.T) {
	client, err := ArgoClientFromConfig(awsArgoCfg)
	require.NoError(t, err)
	stub := &wfClientStub{}
	client.wfClient = stub
	wfName := "sampleName"
	_, err = client.Get(context.TODO(), wfName)
	require.NoError(t, err)
	require.Equal(t, wfName, stub.workflowGetRequest.Name)
	require.Equal(t, awsArgoCfg.Namespace, stub.workflowGetRequest.Namespace)
}

func TestArgoClient_Stop(t *testing.T) {
	client, err := ArgoClientFromConfig(awsArgoCfg)
	require.NoError(t, err)
	stub := &wfClientStub{}
	client.wfClient = stub
	wfName := "sampleName"
	_, err = client.Stop(context.TODO(), wfName)
	require.NoError(t, err)
	require.Equal(t, wfName, stub.workflowStopRequest.Name)
	require.Equal(t, awsArgoCfg.Namespace, stub.workflowStopRequest.Namespace)
}