package import_cases

import (
	l4g "github.com/alecthomas/log4go"
	"github.com/tidbcloud/qa-PlatformAPIAuto/models"
	"testing"

	"github.com/dailymotion/allure-go"
	imp "github.com/tidbcloud/qa-PlatformAPIAuto/webScene/import"
)

/* compress_format.go contains basic operation for import test cases.

compress:
1. gzip
2. snappy
3. zstd

cluster type:
1. dedicated cluster in aws
2. dedicated cluster in gcp
3. serverless cluster in aws
*/

type CompressFormatsTest struct {
	imp.BaseImportScene
}

func (t *CompressFormatsTest) Run() {
	allure.Test(
		t.T,
		allure.Epic("TestImportCompressFormatsTest"),
		allure.Feature("CreateImportTillCompleted"),
		allure.Description("1. create snappy parquet; 2. create sql.(gzip/snappy/zstd); "+
			"3. create svc.(gzip/snappy/zstd)"),
		allure.Action(func() {
			t.CreateClusterStep()

			// If the provider is aws, we should create import credential one time in the test
			if t.Provider == imp.AwsProvider {
				t.CreateImportCredentialStep()
			}
			// Each test should delete import credential
			//defer t.DeleteImportCredentialStep()

			//import schema and data from s3
			t.initData(imp.ImportDataFormatParquet, "", "NoTargetTable")
			t.ImportPreCheckStep()
			t.CreateImportTillCompletedStep()

			//import into pre-created tables
			t.initData(imp.ImportDataFormatParquet, "", "TargetTable")
			t.TruncateTargetTableStep(parquetCompressOperationTable)
			t.SearchImportFiles()
			t.ImportPreCheckStep()
			t.CreateImportTillCompletedStep()

			t.CreateTargetTableStep(sqlOperationTable)
			t.initData(imp.ImportDataFormatSql, "gzip", "TargetTable")
			t.SearchImportFiles()
			t.ImportPreCheckStep()
			t.CreateImportTillCompletedStep()
			t.TruncateTargetTableStep(sqlOperationTable)
			t.initData(imp.ImportDataFormatSql, "zstd", "TargetTable")
			t.CreateImportTillCompletedStep()
			t.TruncateTargetTableStep(sqlOperationTable)
			t.initData(imp.ImportDataFormatSql, "snappy", "TargetTable")
			t.CreateImportTillCompletedStep()

			t.CreateTargetTableStep(csvOperationTable)
			t.initData(imp.ImportDataFormatCsv, "gzip", "TargetTable")
			t.SearchImportFiles()
			t.ImportPreCheckStep()
			t.CreateImportTillCompletedStep()
			t.TruncateTargetTableStep(csvOperationTable)
			t.initData(imp.ImportDataFormatCsv, "zstd", "TargetTable")
			t.CreateImportTillCompletedStep()
			t.TruncateTargetTableStep(csvOperationTable)
			t.initData(imp.ImportDataFormatCsv, "snappy", "TargetTable")
			t.CreateImportTillCompletedStep()
		}),
	)
}

var sqlOperationTable = imp.Table{
	SchemaName:      "dbaas",
	TableName:       "person",
	SchemaStatement: "CREATE TABLE IF NOT EXISTS `dbaas`.`person` (\n  `id` int DEFAULT NULL,\n  `name` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL,\n  `birthday` date DEFAULT NULL\n) ENGINE=InnoDB DEFAULT CHARSET=utf8;",
}

var csvOperationTable = imp.Table{
	SchemaName:      "test",
	TableName:       "t1",
	SchemaStatement: "CREATE TABLE IF NOT EXISTS `test`.`t1` (\n  `a` int(11) DEFAULT NULL,\n   `b` int(11) GENERATED ALWAYS AS (`a` + 1) VIRTUAL,\n  KEY `idx_v_b` (`b`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;\n",
}

var parquetCompressOperationTable = imp.Table{
	SchemaName:      "testdata",
	TableName:       "item",
	SchemaStatement: "CREATE TABLE IF NOT EXISTS `testdata`.`item`  (\n  `i_id` int(11) NOT NULL,\n  `i_im_id` int(11) DEFAULT NULL,\n  `i_name` varchar(24) DEFAULT NULL,\n  `i_price` decimal(5,2) DEFAULT NULL,\n  `i_data` varchar(50) DEFAULT NULL,\n  PRIMARY KEY (`i_id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;",
}

var parquetEmptyTable = imp.Table{
	SchemaName:      "account",
	TableName:       "audits_log",
	SchemaStatement: "create table `account`.`audits_log` (`id` int primary key);",
}

var parquetOperationTable = imp.Table{
	SchemaName:      "import",
	TableName:       "jxtest",
	SchemaStatement: "",
}

// RunImportTestBasicOperationInDedicatedAws is a test case for
// basic import operations, like create import, cancel import, in dedicated aws.
func RunImportTestCompressFormats(t *testing.T, clusterId, provider string, isServerless bool) {
	test := &CompressFormatsTest{}

	test.Init(t, clusterId, provider, isServerless)
	test.Run()
}

// initData is used to init import data for different providers
// data should be created previously now, maybe generated by code is better later
func (t *CompressFormatsTest) initData(dataFormat imp.ImportDataFormat, compress, model string) {
	var importData *models.ClusterIdImportsBody
	var searchImportFile *models.ImportsSearchimportfilesBody
	var precheckData *models.ImportsValidateBody
	var url, schema, table string
	var csv *models.DataflowCustomCsvFormat
	if t.Provider == imp.AwsProvider {
		if dataFormat == imp.ImportDataFormatSql {
			if compress == "gzip" {
				url = "s3://import-scene-test/compress-format/sql-gzip/"
			} else if compress == "zstd" {
				url = "s3://import-scene-test/compress-format/sql-zstd/"
			} else if compress == "snappy" {
				url = "s3://import-scene-test/compress-format/sql-snappy/"
			}
			schema = sqlOperationTable.SchemaName
			table = sqlOperationTable.TableName
		} else if dataFormat == imp.ImportDataFormatCsv {
			csv = &models.DataflowCustomCsvFormat{BackslashEscape: true, Delimiter: "\"", Header: true, Null: "\\N", Separator: ","}
			if compress == "gzip" {
				url = "s3://import-scene-test/compress-format/csv-gzip/"
			} else if compress == "zstd" {
				url = "s3://import-scene-test/compress-format/csv-zstd/"
			} else if compress == "snappy" {
				url = "s3://import-scene-test/compress-format/csv-snappy/"
			}
			schema = csvOperationTable.SchemaName
			table = csvOperationTable.TableName
		} else if dataFormat == imp.ImportDataFormatParquet {
			url = "s3://import-scene-test/compress-format/parquet-snappy/"
			schema = parquetCompressOperationTable.SchemaName
			table = parquetCompressOperationTable.TableName
		}
		arn := t.ImportArnName()
		region := "ap-southeast-1"
		importData = &models.ClusterIdImportsBody{
			SourceUrl:      url,
			DataSourceType: int64(imp.ImportDataSourceTypeS3),
			Region:         region,
			DataFormat:     int64(dataFormat),
			AwsRoleArn:     arn,
			CsvFormat:      csv,
		}
		precheckData = &models.ImportsValidateBody{
			AwsRoleArn:     arn,
			DataFormat:     int64(dataFormat),
			DataSourceType: int64(imp.ImportDataSourceTypeS3),
			Region:         region,
			SourceUrl:      url,
			CheckItems:     []models.DataflowImportValidationItem{models.IMPORT_PRECHECK_DataflowImportValidationItem},
			CsvFormat:      csv,
		}
		searchImportFile = &models.ImportsSearchimportfilesBody{
			AwsRoleArn:     arn,
			DataFormat:     int64(dataFormat),
			DataSourceType: int64(imp.ImportDataSourceTypeS3),
			SourceUri:      url,
		}
	} else {
		if dataFormat == imp.ImportDataFormatSql {
			if compress == "gzip" {
				url = "gs://import-scene-test/compress-format/sql-gzip/"
			} else if compress == "zstd" {
				url = "gs://import-scene-test/compress-format/sql-zstd/"
			} else if compress == "snappy" {
				url = "gs://import-scene-test/compress-format/sql-snappy/"
			}
			schema = sqlOperationTable.SchemaName
			table = sqlOperationTable.TableName
		} else if dataFormat == imp.ImportDataFormatCsv {
			csv = &models.DataflowCustomCsvFormat{BackslashEscape: true, Delimiter: "\"", Header: true, Null: "\\N", Separator: ","}
			if compress == "gzip" {
				url = "gs://import-scene-test/compress-format/csv-gzip/"
			} else if compress == "zstd" {
				url = "gs://import-scene-test/compress-format/csv-zstd/"
			} else if compress == "snappy" {
				url = "gs://import-scene-test/compress-format/csv-snappy/"
			}
			schema = csvOperationTable.SchemaName
			table = csvOperationTable.TableName
		} else if dataFormat == imp.ImportDataFormatParquet {
			url = "gs://import-scene-test/compress-format/parquet-snappy/"
			schema = parquetCompressOperationTable.SchemaName
			table = parquetCompressOperationTable.TableName
		}
		importData = &models.ClusterIdImportsBody{
			SourceUrl:      url,
			DataSourceType: int64(imp.ImportDataSourceTypeGcs),
			DataFormat:     int64(dataFormat),
			CsvFormat:      csv,
		}
		precheckData = &models.ImportsValidateBody{
			DataFormat:     int64(dataFormat),
			DataSourceType: int64(imp.ImportDataSourceTypeGcs),
			SourceUrl:      url,
			CheckItems:     []models.DataflowImportValidationItem{models.IMPORT_PRECHECK_DataflowImportValidationItem},
			CsvFormat:      csv,
		}
		searchImportFile = &models.ImportsSearchimportfilesBody{
			DataFormat:     int64(dataFormat),
			DataSourceType: int64(imp.ImportDataSourceTypeGcs),
			SourceUri:      url,
		}
	}

	targetTable := &models.CommonTable{
		Schema: schema,
		Table:  table,
	}
	targetTableInfo.TargetTable = targetTable
	targetTableInfo.CustomFile = ""
	l4g.Info("targetTable:%v", targetTableInfo)

	if model == "TargetTable" {
		importData.TargetTableInfos = []models.DataflowImportTargetTableInfo{targetTableInfo}
		precheckData.TargetTableInfos = []models.DataflowImportTargetTableInfo{targetTableInfo}
		searchImportFile.TargetTableInfos = []models.DataflowImportTargetTableInfo{targetTableInfo}
	}
	if t.Provider == imp.GcpProvider {
		t.DeleteCredential()
	}
	t.SetImportData(importData)
	t.SetPrecheckData(precheckData)
	t.SetSearchFileData(searchImportFile)
	// If the provider is gcp, we should create import credential each init new data set
	if t.Provider == imp.GcpProvider {
		t.CreateImportCredentialStep()
	}
}