Add e2e import

cli-test/go.mod

@@ -0,0 +1,41 @@
+module cli-test
+
+go 1.22
+
+require (
+	github.com/go-sql-driver/mysql v1.8.1
+	github.com/tidbcloud/pkgv2 v0.4.45
+	github.com/tidbcloud/tidbcloud-cli v1.0.0-nightly-20240806.0.20240913103411-5b207ae472bd
+	go.uber.org/zap v1.27.0
+)
+
+require (
+	filippo.io/edwards25519 v1.1.0 // indirect
+	github.com/gin-contrib/sse v0.1.0 // indirect
+	github.com/gin-gonic/gin v1.7.4 // indirect
+	github.com/go-playground/locales v0.13.0 // indirect
+	github.com/go-playground/universal-translator v0.17.0 // indirect
+	github.com/go-playground/validator/v10 v10.4.1 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/leodido/go-urn v1.2.0 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pingcap/log v1.1.0 // indirect
+	github.com/rs/xid v1.3.0 // indirect
+	github.com/ugorji/go/codec v1.1.7 // indirect
+	go.opentelemetry.io/otel v1.21.0 // indirect
+	go.opentelemetry.io/otel/trace v1.21.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f // indirect
+	google.golang.org/grpc v1.59.0 // indirect
+	google.golang.org/protobuf v1.34.1 // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+)

cli-test/import/e2e_azure.go

@@ -0,0 +1,101 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"os"
+	"os/exec"
+	"regexp"
+	"time"
+
+	"github.com/tidbcloud/pkgv2/log"
+	"go.uber.org/zap"
+)
+
+func e2eAzureImport(ctx context.Context, clusterID string, db *sql.DB) {
+	logger := log.WithContextL(ctx).With(zap.String("test", "e2eAzureImport"))
+	_, err := db.Exec("DROP TABLE IF EXISTS a")
+	if err != nil {
+		logger.Fatal("failed to drop table -> ", zap.Error(err))
+	}
+
+	logger.Info("start import")
+	startImportContext, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+
+	cmd := exec.CommandContext(
+		startImportContext,
+		ticloudCmd,
+		"serverless",
+		"import",
+		"start",
+		"-c", clusterID,
+		"--source-type", "AZURE_BLOB",
+		"--azblob.sas-token", os.Getenv(EnvAzureSaToken),
+		"--azblob.uri", "azure://yxstudent.blob.core.windows.net/student/",
+		"--file-type", "CSV",
+		"--csv.separator", ";")
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	err = cmd.Run()
+	if err != nil {
+		logger.Fatal("failed to start import -> ", zap.Error(err), zap.String("stdout", stdout.String()), zap.String("stderr", stderr.String()))
+	}
+
+	re := regexp.MustCompile(importIDRegexp)
+	matched := re.FindStringSubmatch(stdout.String())
+	if len(matched) != 2 {
+		logger.Fatal("failed to parse import task ID")
+	}
+	importID := matched[1]
+	err = waitImport(ctx, clusterID, importID, logger)
+	if err != nil {
+		logger.Fatal("import failed -> ", zap.Error(err))
+	}
+	logger.Info("import finished")
+}
+
+func e2eAzureNoPrivilegeImport(ctx context.Context, clusterID string, db *sql.DB) {
+	logger := log.WithContextL(ctx).With(zap.String("test", "e2eAzureImport"))
+	_, err := db.Exec("DROP TABLE IF EXISTS a")
+	if err != nil {
+		logger.Fatal("failed to drop table -> ", zap.Error(err))
+	}
+
+	logger.Info("start import")
+	startImportContext, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+
+	cmd := exec.CommandContext(
+		startImportContext,
+		ticloudCmd,
+		"serverless",
+		"import",
+		"start",
+		"-c", clusterID,
+		"--source-type", "AZURE_BLOB",
+		"--azblob.sas-token", os.Getenv(EnvAzureSaTokenNoPrivilege),
+		"--azblob.uri", "azure://yxstudent.blob.core.windows.net/student/",
+		"--file-type", "CSV",
+		"--csv.separator", ";")
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	err = cmd.Run()
+	if err != nil {
+		logger.Fatal("failed to start import -> ", zap.Error(err), zap.String("stdout", stdout.String()), zap.String("stderr", stderr.String()))
+	}
+
+	re := regexp.MustCompile(importIDRegexp)
+	matched := re.FindStringSubmatch(stdout.String())
+	if len(matched) != 2 {
+		logger.Fatal("failed to parse import task ID")
+	}
+	importID := matched[1]
+	err = waitImport(ctx, clusterID, importID, logger)
+	expectFail(err, "error", logger)
+}

cli-test/import/e2e_file_type.go

@@ -0,0 +1,187 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"os"
+	"os/exec"
+	"regexp"
+	"time"
+
+	"github.com/tidbcloud/pkgv2/log"
+	"go.uber.org/zap"
+)
+
+func e2eParquetImport(ctx context.Context, clusterID string, db *sql.DB) {
+	logger := log.WithContextL(ctx).With(zap.String("test", "e2eParquetImport"))
+	_, err := db.Exec("DROP TABLE IF EXISTS ppp")
+	if err != nil {
+		logger.Fatal("failed to drop table -> ", zap.Error(err))
+	}
+
+	logger.Info("start import")
+	startImportContext, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+
+	cmd := exec.CommandContext(
+		startImportContext,
+		ticloudCmd,
+		"serverless",
+		"import",
+		"start",
+		"-c", clusterID,
+		"--s3.uri", "s3://tidbcloud-samples-testimport/yx-test-import/test-parquet/",
+		"--s3.role-arn", os.Getenv(EnvAwsRoleArn),
+		"--source-type", "S3",
+		"--file-type", "PARQUET")
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	err = cmd.Run()
+	if err != nil {
+		logger.Fatal("failed to start import -> ", zap.Error(err), zap.String("stdout", stdout.String()), zap.String("stderr", stderr.String()))
+	}
+
+	re := regexp.MustCompile(importIDRegexp)
+	matched := re.FindStringSubmatch(stdout.String())
+	if len(matched) != 2 {
+		logger.Fatal("failed to parse import task ID")
+	}
+	importID := matched[1]
+	err = waitImport(ctx, clusterID, importID, logger)
+	if err != nil {
+		logger.Fatal("import failed -> ", zap.Error(err))
+	}
+	logger.Info("import finished")
+}
+
+func e2eSchemaCompressImport(ctx context.Context, clusterID string, db *sql.DB) {
+	logger := log.WithContextL(ctx).With(zap.String("test", "e2eSchemaCompressImport"))
+	_, err := db.Exec("DROP TABLE IF EXISTS a")
+	if err != nil {
+		logger.Fatal("failed to drop table -> ", zap.Error(err))
+	}
+
+	logger.Info("start import")
+	startImportContext, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+
+	cmd := exec.CommandContext(
+		startImportContext,
+		ticloudCmd,
+		"serverless",
+		"import",
+		"start",
+		"-c", clusterID,
+		"--s3.uri", "s3://tidbcloud-samples-testimport/yx-test-import/test-with-schema-compress/",
+		"--s3.role-arn", os.Getenv(EnvAwsRoleArn),
+		"--source-type", os.Getenv(EnvAwsRoleArn),
+		"--file-type", "CSV",
+		"--csv.separator", ";")
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	err = cmd.Run()
+	if err != nil {
+		logger.Fatal("failed to start import -> ", zap.Error(err), zap.String("stdout", stdout.String()), zap.String("stderr", stderr.String()))
+	}
+
+	re := regexp.MustCompile(importIDRegexp)
+	matched := re.FindStringSubmatch(stdout.String())
+	if len(matched) != 2 {
+		logger.Fatal("failed to parse import task ID")
+	}
+	importID := matched[1]
+	err = waitImport(ctx, clusterID, importID, logger)
+	if err != nil {
+		logger.Fatal("import failed -> ", zap.Error(err))
+	}
+	logger.Info("import finished")
+}
+
+func e2eSchemaTypeMisMatchedImport(ctx context.Context, clusterID string, db *sql.DB) {
+	logger := log.WithContextL(ctx).With(zap.String("test", "e2eSchemaTypeMisMatchedImport"))
+	_, err := db.Exec("DROP TABLE IF EXISTS a")
+	if err != nil {
+		logger.Fatal("failed to drop table -> ", zap.Error(err))
+	}
+
+	logger.Info("start import")
+	startImportContext, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+
+	cmd := exec.CommandContext(
+		startImportContext,
+		ticloudCmd,
+		"serverless",
+		"import",
+		"start",
+		"-c", clusterID,
+		"--s3.uri", "s3://tidbcloud-samples-testimport/yx-test-import/test-schema-type-mismatch/",
+		"--s3.role-arn", os.Getenv(EnvAwsRoleArn),
+		"--source-type", "S3",
+		"--file-type", "CSV",
+		"--csv.separator", ";")
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	err = cmd.Run()
+	if err != nil {
+		logger.Fatal("failed to start import -> ", zap.Error(err), zap.String("stdout", stdout.String()), zap.String("stderr", stderr.String()))
+	}
+
+	re := regexp.MustCompile(importIDRegexp)
+	matched := re.FindStringSubmatch(stdout.String())
+	if len(matched) != 2 {
+		logger.Fatal("failed to parse import task ID")
+	}
+	importID := matched[1]
+	err = waitImport(ctx, clusterID, importID, logger)
+	expectFail(err, "failed to cast value as int(11) for column `name`", logger)
+}
+
+func e2eSchemaColumnNumberMismatchImport(ctx context.Context, clusterID string, db *sql.DB) {
+	logger := log.WithContextL(ctx).With(zap.String("test", "e2eSchemaColumnNumberMismatchImport"))
+	_, err := db.Exec("DROP TABLE IF EXISTS a")
+	if err != nil {
+		logger.Fatal("failed to drop table -> ", zap.Error(err))
+	}
+
+	logger.Info("start import")
+	startImportContext, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+
+	cmd := exec.CommandContext(
+		startImportContext,
+		ticloudCmd,
+		"serverless",
+		"import",
+		"start",
+		"-c", clusterID,
+		"--s3.uri", "s3://tidbcloud-samples-testimport/yx-test-import/test-schema-column-number-mismatch/",
+		"--s3.role-arn", os.Getenv(EnvAwsRoleArn),
+		"--source-type", "S3",
+		"--file-type", "CSV",
+		"--csv.separator", ";")
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	err = cmd.Run()
+	if err != nil {
+		logger.Fatal("failed to start import -> ", zap.Error(err), zap.String("stdout", stdout.String()), zap.String("stderr", stderr.String()))
+	}
+
+	re := regexp.MustCompile(importIDRegexp)
+	matched := re.FindStringSubmatch(stdout.String())
+	if len(matched) != 2 {
+		logger.Fatal("failed to parse import task ID")
+	}
+	importID := matched[1]
+	err = waitImport(ctx, clusterID, importID, logger)
+	expectFail(err, "TiDB schema `test`.`a` doesn't have the default value for number", logger)
+}