// +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"
	"fmt"
	"os"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"gorm.io/gorm"
)

func buildJobDao(assert *require.Assertions) (JobDao, *gorm.DB, error) {
	cfg := &Config{
		Driver: "mysql",
		DSN:    os.Getenv("TIDB_DSN"),
	}
	cfg.FillWithDefaults()
	db, err := Dial(cfg)
	if err != nil {
		return nil, nil, err
	}
	assert.Nil(db.Exec("drop table if exists migrations,job,savepoint").Error)
	assert.Nil(Migrate(db))

	dao, err := NewJobDao(WithDB(db))
	if err != nil {
		return nil, nil, err
	}
	return dao, db, nil
}

func TestNewJobDao(t *testing.T) {
	assert := require.New(t)
	cfg := &Config{
		Driver: "mysql",
		DSN:    os.Getenv("TIDB_DSN"),
	}
	cfg.FillWithDefaults()
	db, err := Dial(cfg)
	assert.Nil(err)
	assert.Nil(db.Exec("drop table if exists migrations,job,savepoint").Error)
	assert.Nil(Migrate(db))

	_, err = NewJobDao(WithDB(db))
	assert.Nil(err)

	assert.Nil(db.Exec("drop table if exists migrations,job,savepoint").Error)
}

func TestJobDao_Insert(t *testing.T) {
	assert := require.New(t)
	dao, db, err := buildJobDao(assert)
	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",
	}
	err = dao.Insert(ctx, &job)
	assert.Nil(err)

	assert.Nil(db.Exec("drop table if exists migrations,job,savepoint").Error)
}

func TestMigrate(t *testing.T) {
	const createJobSQL = "CREATE TABLE `job` (\n  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,\n  `user_id` varchar(64) NOT NULL DEFAULT '',\n  `job_id` varchar(64) NOT NULL DEFAULT '',\n  `jm_endpoint` varchar(128) NOT NULL DEFAULT '',\n  `namespace` varchar(64) NOT NULL DEFAULT '',\n  `job_type` tinyint(4) NOT NULL DEFAULT '0',\n  `input_type` tinyint(4) NOT NULL DEFAULT '0',\n  `input_key` varchar(128) NOT NULL DEFAULT '',\n  `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,\n  `updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,\n  PRIMARY KEY (`id`) /*T![clustered_index] CLUSTERED */,\n  UNIQUE KEY `job_user_id_job_id_index` (`job_id`,`user_id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"
	const createSavepointSQL = "CREATE TABLE `savepoint` (\n  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,\n  `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,\n  `updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,\n  `job_ref` int(11) NOT NULL,\n  `path` text DEFAULT NULL,\n  PRIMARY KEY (`id`) /*T![clustered_index] CLUSTERED */,\n  CONSTRAINT `fk_savepoint_job` FOREIGN KEY (`job_ref`) REFERENCES `job` (`id`) ON DELETE CASCADE\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"
	assert := require.New(t)
	cfg := &Config{
		Driver: "mysql",
		DSN:    os.Getenv("TIDB_DSN"),
	}
	cfg.FillWithDefaults()
	db, err := Dial(cfg)
	assert.Nil(err)
	assert.Nil(db.Exec("drop table if exists migrations,job,savepoint").Error)
	assert.Nil(Migrate(db))
	{
		var table string
		var cTable string
		err = db.Raw("show create table job").Row().Scan(&table, &cTable)
		assert.Nil(err)
		assert.Equal(createJobSQL, cTable)
	}
	{
		var table string
		var cTable string
		err = db.Raw("show create table savepoint").Row().Scan(&table, &cTable)
		assert.Nil(err)
		assert.Equal(createSavepointSQL, cTable)
	}
	jobDao, err := NewJobDao(WithDB(db))
	assert.Nil(err)
	spDao, err := NewSavepointDao(WithDB(db))
	assert.Nil(err)
	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",
	}
	assert.Nil(jobDao.Insert(ctx, &job))

	sp := Savepoint{
		JobRef: job.ID,
		Path:   "/tmp",
	}
	assert.Nil(spDao.Insert(ctx, &sp))

	assert.Nil(db.Exec("drop table if exists migrations,job,savepoint").Error)
}