package pimport
import (
"fmt"
"testing"
"github.com/pingcap/errcode"
"github.com/pingcap/errors"
"github.com/stretchr/testify/require"
agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
"github.com/tidbcloud/dataflow-service/internal/db/entity"
pimport "github.com/tidbcloud/dataflow-service/internal/model/import"
"go.uber.org/zap"
)
func TestTranslateLightningPrecheckAgentError(t *testing.T) {
type testCase struct {
InputErr error
DataFormat entity.DataFormat
ExpectedOutputErrStr string
}
const (
testAwsRoleARN = "arn:aws:iam::888888888888:role/some-role"
testAwsSource = "s3://some-bucket/some-folder"
)
testCases := []*testCase{
{
InputErr: errors.New("run precheck item 'CHECK_ITEM_SOURCE_SCHEMA' error: generate table info from SQL error: sbtest1: line 6 column 14 near \"(`id`) /*T![clustered_index] CLUSTERED */,\nUNIQUE KEY `some_index` (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_bin;\""),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: "The table 'sbtest1' has some problems with the schema file: line 6 column 14 near \"(`id`) /*T![clustered_index] CLUSTERED */,\nUNIQUE KEY `some_index` (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_bin;\". Please check the content of the schema file.",
},
{
InputErr: errors.New("generate precheck builder from config error: [Lightning:Common:ErrUnknown]unknown error: s3 bucket and region are not matched, bucket=tidbcloud-samples, input region=us-east-1, real region=us-west-2"),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: "The provided region of S3 bucket 'tidbcloud-samples' is not correct: provided region is 'us-east-1'; actual region is 'us-west-2'",
},
{
InputErr: errors.Errorf("generate precheck builder from config error: [Lightning:Storage:ErrStorageUnknown]list file failed: AccessDenied: User: arn:aws:sts::999999999999:assumed-role/prod-abcdefg/gfedcba is not authorized to perform: sts:AssumeRole on resource: %s\n\tstatus code: 403, request id: 00000000-0000-0000-0000-000000000000", testAwsRoleARN),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: assumeRoleErr(testAwsRoleARN).Error(),
},
{
InputErr: errors.New("run precheck item 'CHECK_ITEM_SOURCE_SCHEMA' error: get create table statement from schema file error: test1: schema file is missing for the table 'db1.test1'"),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: "Cannot find the table schema file for the table 'db1.test1'. Please make sure the table schema file is put in the data source dir, and it has the correct file name.",
},
{
InputErr: errors.New("run precheck item 'CHECK_ITEM_SOURCE_SCHEMA' error: [Lightning:Restore:ErrSchemaNotExists]table `db01`.`tbl01` schema not found"),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: "Cannot find the create statement for the table `db01`.`tbl01`. Please check the create table statement in the schema file, and make it consistent with the table name the file refers to.",
},
{
InputErr: errors.New("generate precheck builder from config error: Error 1045: Access denied for user 'cloud_admin'@'10.0.1.2' (using password: YES)"),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: "Access denied to the TiDB cluster temporarily. Please try again later.",
},
{
InputErr: errors.New("run precheck item 'CHECK_ITEM_SOURCE_SCHEMA' error: syntax error: cannot have consecutive fields without separator"),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: "Syntax errors: The CSV source file cannot be parsed with the current 'CSV configuration'. Please modify the 'CSV Configuration' settings to ensure correct parsing of the CSV source file.",
},
{
InputErr: errors.New(`Post "https://some-url.example.com/api/v1/import/lightning-precheck": context deadline exceeded (Client.Timeout exceeded while awaiting headers)`),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: unidentifiedError.Error(),
},
{
InputErr: errors.New(`Post "https://some-url.example.com/api/v1/import/lightning-precheck": context deadline exceeded (Client.Timeout exceeded while awaiting headers)`),
DataFormat: entity.ImportDataFormatAuroraSnapshot,
ExpectedOutputErrStr: "",
},
{
InputErr: errors.New(`Post "https://some-url.example.com/api/v1/import/lightning-precheck": EOF`),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: unidentifiedError.Error(),
},
{
InputErr: errors.New(`Post "https://some-url.example.com/api/v1/import/lightning-precheck": EOF`),
DataFormat: entity.ImportDataFormatAuroraSnapshot,
ExpectedOutputErrStr: "",
},
{
InputErr: errors.New("run precheck item 'CHECK_ITEM_SOURCE_SCHEMA' error: gzip: invalid header"),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: "Unable to extract the compressed data file. The compression algorithm used in the file does not match the file compression suffix. Please ensure that the file was compressed using the correct algorithm and try again.",
},
{
InputErr: errors.New("abcd1234"),
DataFormat: entity.ImportDataFormatCsv,
ExpectedOutputErrStr: unidentifiedError.Error(),
},
}
for _, tc := range testCases {
outputErr := translateLightningPrecheckAgentError(tc.InputErr, entity.ImportDataSourceS3, tc.DataFormat, "", "", testAwsRoleARN, testAwsSource, zap.L())
if len(tc.ExpectedOutputErrStr) > 0 {
require.EqualError(t, outputErr, tc.ExpectedOutputErrStr)
} else {
require.NoError(t, outputErr)
}
}
}
func TestGeneratePrecheckError(t *testing.T) {
type testCase struct {
CheckResult *agentpb.CheckResult
ExpectedOutputErrStr string
}
testCases := []*testCase{
{
CheckResult: &agentpb.CheckResult{
CheckItem: agentpb.CheckItem_CHECK_ITEM_SOURCE_SCHEMA,
HasPassed: false,
Msg: "TiDB schema `test`.`warehouse` doesn't have column 1, please check table schema and csv file header or use tables.ignoreColumns to ignore 1\nTiDB schema `test`.`warehouse` doesn't have column sfb5cewv, please check table schema and csv file header or use tables.ignoreColumns to ignore sfb5cewv\nTiDB schema `test`.`warehouse` doesn't have column c6d7cbgycyeje3w, please check table schema and csv file header or use tables.ignoreColumns to ignore c6d7cbgycyeje3w\nTiDB schema `test`.`warehouse` doesn't have column 5ol67lrlgd, please check table schema and csv file header or use tables.ignoreColumns to ignore 5ol67lrlgd\nTiDB schema `test`.`warehouse` doesn't have column v4zgxjvl1mzb, please check table schema and csv file header or use tables.ignoreColumns to ignore v4zgxjvl1mzb\nTiDB schema `test`.`warehouse` doesn't have column bi, please check table schema and csv file header or use tables.ignoreColumns to ignore bi\nTiDB schema `test`.`warehouse` doesn't have column 731611111, please check table schema and csv file header or use tables.ignoreColumns to ignore 731611111\nTiDB schema `test`.`warehouse` doesn't have column 0.128800, please check table schema and csv file header or use tables.ignoreColumns to ignore 0.128800\nTiDB schema `test`.`warehouse` doesn't have column 300000.000000, please check table schema and csv file header or use tables.ignoreColumns to ignore 300000.000000\nTiDB schema `test`.`warehouse` doesn't have the default value for w_id. Please add default value for column 'w_id' or choose another column to ignore or add this column in data file\nTiDB schema `test`.`district` doesn't have column 1, please check table schema and csv file header or use tables.ignoreColumns to ignore 1\nTiDB schema `test`.`district` doesn't have column 1, please check table schema and csv file header or use tables.ignoreColumns to ignore 1\nTiDB schema `test`.`district` doesn't have column ii0acdald, please check table schema and csv file header or use tables.ignoreColumns to ignore ii0acdald\nTiDB schema `test`.`district` doesn't have column fjgjopypgk, please check table schema and csv file header or use tables.ignoreColumns to ignore fjgjopypgk\nTiDB schema `test`.`district` doesn't have column g4ytuewmgffktojed, please check table schema and csv file header or use tables.ignoreColumns to ignore g4ytuewmgffktojed\nTiDB schema `test`.`district` doesn't have column 4jwhmafwygaiik6g, please check table schema and csv file header or use tables.ignoreColumns to ignore 4jwhmafwygaiik6g\nTiDB schema `test`.`district` doesn't have column qy, please check table schema and csv file header or use tables.ignoreColumns to ignore qy\nTiDB schema `test`.`district` doesn't have column 620411111, please check table schema and csv file header or use tables.ignoreColumns to ignore 620411111\nTiDB schema `test`.`district` doesn't have column 0.062600, please check table schema and csv file header or use tables.ignoreColumns to ignore 0.062600\nTiDB schema `test`.`district` doesn't have column 30000.000000, please check table schema and csv file header or use tables.ignoreColumns to ignore 30000.000000\nTiDB schema `test`.`district` doesn't have column 3001, please check table schema and csv file header or use tables.ignoreColumns to ignore 3001\nTiDB schema `test`.`district` doesn't have the default value for d_w_id. Please add default value for column 'd_w_id' or choose another column to ignore or add this column in data file\nTiDB schema `test`.`district` doesn't have the default value for d_id. Please add default value for column 'd_id' or choose another column to ignore or add this column in data file\nTiDB schema `test`.`item` doesn't have column 1, please check table schema and csv file header or use tables.ignoreColumns to ignore 1\nTiDB schema `test`.`item` doesn't have column 8879, please check table schema and csv file header or use tables.ignoreColumns to ignore 8879",
},
ExpectedOutputErrStr: "Souce data is inconsistent with the table schema: affected tables: [`test`.`district`,`test`.`item`,`test`.`warehouse`]; error details: TiDB schema `test`.`warehouse` doesn't have column 1, please check table schema and csv file header or use tables.ignoreColumns to ignore 1; TiDB schema `test`.`warehouse` doesn't have column sfb5cewv, please check table schema and csv file header or use tables.ignoreColumns to ignore sfb5cewv; TiDB schema `test`.`warehouse` doesn't have column c6d7cbgycyeje3w, please check table schema and csv file header or use tables.ignoreColumns to ignore c6d7cbgycyeje3w ... and 22 more errors. Please try to fix the source data and try again",
},
{
CheckResult: &agentpb.CheckResult{
CheckItem: agentpb.CheckItem_CHECK_ITEM_SOURCE_SCHEMA,
HasPassed: false,
Msg: "TiDB schema `test`.`warehouse` doesn't have column 1, please check table schema and csv file header or use tables.ignoreColumns to ignore 1\nTiDB schema `test`.`warehouse` doesn't have column sfb5cewv, please check table schema and csv file header or use tables.ignoreColumns to ignore sfb5cewv",
},
ExpectedOutputErrStr: "Souce data is inconsistent with the table schema: affected tables: [`test`.`warehouse`]; error details: TiDB schema `test`.`warehouse` doesn't have column 1, please check table schema and csv file header or use tables.ignoreColumns to ignore 1; TiDB schema `test`.`warehouse` doesn't have column sfb5cewv, please check table schema and csv file header or use tables.ignoreColumns to ignore sfb5cewv. Please try to fix the source data and try again",
},
}
for _, tc := range testCases {
outputErr := generatePrecheckError(tc.CheckResult)
require.EqualError(t, outputErr, tc.ExpectedOutputErrStr)
}
}
func TestTranslatePreviewAgentError(t *testing.T) {
type testCase struct {
InputErr error
ExpectedOutputErrStr string
ExpectedHasTransformed bool
}
testCases := []*testCase{
{
InputErr: errors.New("preview source error: generate table info from SQL error: sbtest1: line 6 column 14 near \"(`id`) /*T![clustered_index] CLUSTERED */,\nUNIQUE KEY `some_index` (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_bin;\""),
ExpectedOutputErrStr: "The table 'sbtest1' has some problems with the schema file: line 6 column 14 near \"(`id`) /*T![clustered_index] CLUSTERED */,\nUNIQUE KEY `some_index` (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_bin;\". Please check the content of the schema file.",
ExpectedHasTransformed: true,
},
{
InputErr: errors.New("preview source error: get create table statement from schema file error: test1: schema file is missing for the table 'db1.test1'"),
ExpectedOutputErrStr: "Cannot find the table schema file for the table 'db1.test1'. Please make sure the table schema file is put in the data source dir, and it has the correct file name.",
ExpectedHasTransformed: true,
},
{
InputErr: errors.New("preview source error: syntax error: cannot have consecutive fields without separator"),
ExpectedOutputErrStr: "Syntax errors: The CSV source file cannot be parsed with the current 'CSV configuration'. Please modify the 'CSV Configuration' settings to ensure correct parsing of the CSV source file.",
ExpectedHasTransformed: true,
},
{
InputErr: errors.New("preview source error: gzip: invalid header"),
ExpectedOutputErrStr: "Unable to extract the compressed data file. The compression algorithm used in the file does not match the file compression suffix. Please ensure that the file was compressed using the correct algorithm and try again.",
ExpectedHasTransformed: true,
},
{
InputErr: errors.New("abcd1234"),
ExpectedOutputErrStr: "abcd1234",
ExpectedHasTransformed: false,
},
}
for _, tc := range testCases {
hasTransformed, outputErr := translatePreviewAgentError(tc.InputErr)
require.Equal(t, tc.ExpectedHasTransformed, hasTransformed)
if len(tc.ExpectedOutputErrStr) > 0 {
require.EqualError(t, outputErr, tc.ExpectedOutputErrStr)
} else {
require.NoError(t, outputErr)
}
}
}
func TestAdjustS3AccessErr(t *testing.T) {
type testCase struct {
InputErrMsg string
ExpectedOutputErrStr string
}
const (
testRoleARN = "arn:aws:iam::XXXXXXXXXXXX:role/my-role"
testAwsAccessKeyId = "AXXXXXXXXXXXXK"
testAwsSecretAccessKey = "SKXXXXXXX"
testS3Path = "s3://mybucket/mydir001"
)
testCases := []*testCase{
{
InputErrMsg: fmt.Sprintf("AccessDenied: User: arn:aws:sts::XXXXXXXXXXXX:assumed-role/xxx/yyy is not authorized to perform: sts:AssumeRole on resource: %s\n\tstatus code: 403, request id: 6562b995-7086-4f16-a7bc-53d634d82f27", testRoleARN),
ExpectedOutputErrStr: assumeRoleErr(testRoleARN).Error(),
},
{
InputErrMsg: "there is no file in this bucket",
ExpectedOutputErrStr: sourceNotFoundErr(testS3Path).Error(),
},
{
InputErrMsg: "list objects error: AccessDenied: Access Denied\n\tstatus code: 403, request id: NMGZYCS75PAXKNN4, host id: NABzTrjFYt/kuJx77v9kkRQoswH4b2CRmAh9WpUjNVNb1wfSifPr61hb/+b5+r9t8TguYUmni6M=",
ExpectedOutputErrStr: s3AccessDeniedErr("", "", testRoleARN, testS3Path, "ListBucket").Error(),
},
{
InputErrMsg: "get object error: AccessDenied: Access Denied\n\tstatus code: 403, request id: NMGZYCS75PAXKNN4, host id: NABzTrjFYt/kuJx77v9kkRQoswH4b2CRmAh9WpUjNVNb1wfSifPr61hb/+b5+r9t8TguYUmni6M=",
ExpectedOutputErrStr: s3AccessDeniedErr("", "", testRoleARN, testS3Path, "GetObject").Error(),
},
{
InputErrMsg: "abcd1234",
ExpectedOutputErrStr: s3AccessDeniedErr("", "", testRoleARN, testS3Path, "").Error(),
},
}
for _, tc := range testCases {
outputErr := adjustS3AccessErr("", "", testRoleARN, testS3Path, tc.InputErrMsg)
if len(tc.ExpectedOutputErrStr) > 0 {
require.EqualError(t, outputErr, tc.ExpectedOutputErrStr, tc.InputErrMsg)
} else {
require.NoError(t, outputErr)
}
}
}
func TestAdjustGCSAccessErr(t *testing.T) {
type testCase struct {
InputErrMsg string
ExpectedOutputErrStr string
}
testGCSPath := "gs://mybucket/mydir001"
testCases := []*testCase{
{
InputErrMsg: "there is no file in this bucket",
ExpectedOutputErrStr: sourceNotFoundErr(testGCSPath).Error(),
},
{
InputErrMsg: "googleapi: Error 403: XXXX does not have storage.buckets.get access to the Google Cloud Storage bucket., forbidden",
ExpectedOutputErrStr: gcsAccessDeniedErr(testGCSPath, "storage.buckets.get").Error(),
},
{
InputErrMsg: "googleapi: Error 403: XXXX does not have storage.objects.list access to the Google Cloud Storage bucket., forbidden",
ExpectedOutputErrStr: gcsAccessDeniedErr(testGCSPath, "storage.objects.list").Error(),
},
{
InputErrMsg: "googleapi: Error 403: XXXX does not have storage.objects.get access to the Google Cloud Storage bucket., forbidden",
ExpectedOutputErrStr: gcsAccessDeniedErr(testGCSPath, "storage.objects.get").Error(),
},
{
InputErrMsg: "abcd1234",
ExpectedOutputErrStr: gcsAccessDeniedErr(testGCSPath, "").Error(),
},
}
for _, tc := range testCases {
outputErr := adjustGCSAccessErr(testGCSPath, tc.InputErrMsg)
if len(tc.ExpectedOutputErrStr) > 0 {
require.EqualError(t, outputErr, tc.ExpectedOutputErrStr)
} else {
require.NoError(t, outputErr)
}
}
}
func TestS3AccessDeniedErr(t *testing.T) {
testCases := []struct {
roleArn string
accessKeyId string
secretAccessKey string
operation string
expected string
}{
{
roleArn: "arn:aws:iam::XXXXXXXXXXXX:role/my-role",
accessKeyId: "",
secretAccessKey: "",
operation: "",
expected: `Access denied to the source 's3://some-bucket/some-folder' using the role 'arn:aws:iam::XXXXXXXXXXXX:role/my-role'. Please refer to the trouble-shooting guide and configure the role and ensure it can access the source.`,
},
{
roleArn: "",
accessKeyId: "AXXXXXXXXXXXXK",
secretAccessKey: "SKXXXXXXX",
operation: "",
expected: `Access denied to the source 's3://some-bucket/some-folder' using the access key id 'AXXXXXXXXXXXXK' and the secret access key 'SKXXXXXXX'. Please refer to the trouble-shooting guide and configure the ak/sk and ensure it can access the source.`,
},
{
roleArn: "",
accessKeyId: "AXXXXXXXXXXXXK",
secretAccessKey: "SKXXXXXXX",
operation: "ListBucket",
expected: `Access denied to the source 's3://some-bucket/some-folder' using the access key id 'AXXXXXXXXXXXXK' and the secret access key 'SKXXXXXXX': The ak/sk does not have ListBucket permission to the source. Please ensure the ak/sk has all the necessary permissions configured in the 'permissions policies' settings. For more information, please refer to the trouble-shooting guide.`,
},
{
roleArn: "",
accessKeyId: "",
secretAccessKey: "",
operation: "ListBucket",
expected: `Access denied to the source 's3://some-bucket/some-folder': The does not have ListBucket permission to the source. Please ensure the has all the necessary permissions configured in the 'permissions policies' settings. For more information, please refer to the trouble-shooting guide.`,
},
}
sourceUrl := "s3://some-bucket/some-folder"
for _, tc := range testCases {
require.Equal(t, tc.expected, s3AccessDeniedErr(tc.accessKeyId, tc.secretAccessKey, tc.roleArn, sourceUrl, tc.operation).Error())
}
}
func TestTranslateValidateImportError(t *testing.T) {
logger, err := zap.NewDevelopment()
require.NoError(t, err)
sourceURL := "gs://example_bucket/example_folder"
var sourceErr error = &pimport.ValidateImportSingleCheckItemError{
Item: dataflow.ImportValidationItem_SOURCE_ACCESS_CHECK,
Err: &pimport.SourceAccessCheckFailedError{
AWSRoleARN: "",
SourceURL: sourceURL,
AWSAccessKeyID: "",
AWSSecretAccessKey: "",
ErrorMsg: "googleapi: Error 403: SOME_USER does not have storage.buckets.get access to the Google Cloud Storage bucket. Permission 'storage.buckets.get' denied on resource (or it may not exist)., forbidden",
DataSourceType: entity.ImportDataSourceGcs,
},
}
translatedErr := TranslateValidateImportError(sourceErr,
&pimport.ImportTask{
Import: entity.Import{
SourceURL: sourceURL,
},
},
logger)
codedErr := translatedErr.(errcode.ErrorCode)
require.Equal(t, errcode.StateCode, codedErr.Code())
sourceURL = "s3://some_bucket/some_folder"
sourceErr = &pimport.ValidateImportSingleCheckItemError{
Item: dataflow.ImportValidationItem_IMPORT_PRECHECK,
Err: errors.Annotate(
errors.Annotate(
errors.New("run precheck item 'CHECK_ITEM_SOURCE_SCHEMA' error: syntax error: cannot have consecutive fields without separator"),
"lightning precheck from agent has errors",
),
"do lightning precheck returns error",
),
}
translatedErr = TranslateValidateImportError(sourceErr,
&pimport.ImportTask{
Import: entity.Import{
SourceURL: sourceURL,
},
},
logger)
codedErr = translatedErr.(errcode.ErrorCode)
require.Equal(t, errcode.InvalidInputCode, codedErr.Code())
}
func TestTranslateImportDataOnlyCheckError(t *testing.T) {
testCases := []struct {
originalErr error
expectedErr error
}{
{
originalErr: errors.New("internal error"),
expectedErr: errors.New("Internal error on import validation, please feel free to contact support."),
},
{
originalErr: errors.New("Error 1105 (HY000): Cluster is temporarily unavailable. Please try again later."),
expectedErr: errors.New("The TiDB cluster is temporarily unavailable. Please try again later."),
},
{
originalErr: errors.New("syntax error: cannot have consecutive fields without separator"),
expectedErr: errors.New("Syntax errors: The CSV source file cannot be parsed with the current 'CSV configuration'. Please modify the 'CSV Configuration' settings to ensure correct parsing of the CSV source file."),
},
}
for _, tc := range testCases {
require.Equal(t, tc.expectedErr.Error(), translateImportDataOnlyCheckError(tc.originalErr).Error())
}
}