// +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 model

import (
	"context"
	"database/sql"
	"fmt"
	"reflect"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"gopkg.in/DATA-DOG/go-sqlmock.v1"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
)

func buildMockJobDao() (JobDao, sqlmock.Sqlmock, error) {
	db, mock, err := sqlmock.New()
	if err != nil {
		return nil, nil, err
	}
	dao, err := NewJobDao(withMockDBOpt(db))
	if err != nil {
		return nil, nil, err
	}
	return dao, mock, nil
}

func withMockDBOpt(db *sql.DB) DaoOption {
	return func(dao Dao) (err error) {
		daoDB, err := gorm.Open(mysql.New(mysql.Config{
			SkipInitializeWithVersion: true,
			Conn:                      db,
		}), &gorm.Config{})
		if err != nil {
			return err
		}
		dao.SetDB(daoDB)
		return nil
	}
}

var rowHeader = []string{
	"id",
	"user_id",
	"job_id",
	"jm_endpoint",
	"namespace",
	"job_type",
	"input_type",
	"input_key",
	"created_at",
	"updated_at"}

func TestJobDao_MockInsert(t *testing.T) {
	assert := require.New(t)
	dao, mock, err := buildMockJobDao()
	assert.Nil(err)
	assert.NotNil(dao)
	ctx := context.Background()
	job := Job{
		UserID:     "test",
		JobID:      fmt.Sprintf("test-%v", time.Now().UnixNano()),
		JMEndpoint: "localhost:80",
		Namespace:  "test",
		JobType:    JobTypeFlinkStreaming,
		InputType:  InputTypeSQL,
		InputKey:   "uuid",
	}
	mock.ExpectBegin()
	mock.ExpectExec("^INSERT INTO `job`(.+) ON DUPLICATE KEY UPDATE").
		WithArgs(
			job.UserID,
			job.JobID,
			job.JMEndpoint,
			job.Namespace,
			job.JobType,
			job.InputType,
			job.InputKey,
		).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectCommit()
	err = dao.Insert(ctx, &job)
	assert.Nil(err)
	assert.Nil(mock.ExpectationsWereMet())
}

func TestJobDao_MockFind(t *testing.T) {
	assert := require.New(t)
	dao, mock, err := buildMockJobDao()
	assert.Nil(err)
	assert.NotNil(dao)
	ctx := context.Background()
	{
		mock.ExpectQuery("SELECT .+ FROM `job` WHERE user_id = \\? && job_id = \\? ORDER BY `job`.`id` LIMIT 1").
			WithArgs("foo", "foo").
			WillReturnRows(sqlmock.NewRows([]string{"id"}))
		job, find, err := dao.Find(ctx, "foo", "foo")
		assert.Nil(err)
		assert.False(find)
		assert.Nil(job)
		assert.Nil(mock.ExpectationsWereMet())
	}
	{
		rs := sqlmock.NewRows(rowHeader).AddRow(
			"1",
			"foo",
			"foo",
			"localhost:80",
			"default",
			1,
			1,
			"foo",
			time.Now(),
			time.Now())
		mock.ExpectQuery("^SELECT .+ FROM `job` WHERE user_id = \\? && job_id = \\? ORDER BY `job`.`id` LIMIT 1").
			WithArgs("foo", "foo").
			WillReturnRows(rs)
		job, find, err := dao.Find(ctx, "foo", "foo")
		assert.Nil(err)
		assert.True(find)
		assert.NotNil(job)
		assert.Nil(mock.ExpectationsWereMet())
	}
}

func TestJobDao_MockList(t *testing.T) {
	assert := require.New(t)
	dao, mock, err := buildMockJobDao()
	assert.Nil(err)
	assert.NotNil(dao)
	ctx := context.Background()

	rs := sqlmock.NewRows(rowHeader).AddRow(
		"1",
		"foo",
		"foo",
		"localhost:80",
		"default",
		1,
		1,
		"foo",
		time.Now(),
		time.Now())
	cond := &ListJobCond{
		UserID: "foo",
		Limit:  10,
	}
	mock.ExpectQuery("SELECT .+ FROM `job` WHERE user_id = \\? LIMIT 10").
		WithArgs(cond.UserID).
		WillReturnRows(rs)

	jobs, err := dao.List(ctx, cond)
	assert.Nil(err)
	assert.Greater(len(jobs), 0)
	assert.Nil(mock.ExpectationsWereMet())
}

func Test_jobDao_Select(t *testing.T) {
	dsn := "root:zSjmAWCaPl@tcp(127.0.0.1:3306)/mimir?charset=utf8mb4&parseTime=True&loc=Local"
	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
	if err != nil {
		panic(err)
	}
	sinkField := "sink_id"
	sinkId := "testtable"
	type fields struct {
		db *gorm.DB
	}
	type args struct {
		ctx    context.Context
		fields []interface{}
	}
	tests := []struct {
		name    string
		fields  fields
		args    args
		want    *Job
		want1   bool
		wantErr bool
	}{
		{
			name: "test",
			fields: fields{
				db: db,
			},
			args: args{
				ctx:    context.Background(),
				fields: []interface{}{&sinkField, &sinkId},
			},
			want: &Job{
				ID:        1,
				JobID:     "testjob",
				IngressId: "testingress",
				SinkId:    "testsink",
			},
			wantErr: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dao := &jobDao{
				db: tt.fields.db,
			}
			got, got1, err := dao.Select(tt.args.ctx, tt.args.fields...)
			if (err != nil) != tt.wantErr {
				t.Errorf("jobDao.Find() error = %v, wantErr %v", err, tt.wantErr)
				return
			}
			if !reflect.DeepEqual(got, tt.want) {
				t.Errorf("jobDao.Find() got = %v, want %v", got, tt.want)
			}
			if got1 != tt.want1 {
				t.Errorf("jobDao.Find() got1 = %v, want %v", got1, tt.want1)
			}
		})
	}
}