package imports

import (
	"fmt"
	"regexp"
	"testing"

	. "github.com/onsi/gomega"
	"github.com/stretchr/testify/require"
	"k8s.io/utils/pointer"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/pingcap/util/config"
)

// generateCSVSource generates CSV format import source config for testing
func generateCSVSource(sep, delimiter, null *string, header, notNull, backslashEscape, trimLastSeparator *bool) cloudv1.Source {
	return cloudv1.Source{
		Format:    cloudv1.DataFormatCSV,
		CsvFormat: &cloudv1.CsvFormat{Separator: sep, Delimiter: delimiter, Header: header, NotNull: notNull, Null: null, BackslashEscape: backslashEscape, TrimLastSeparator: trimLastSeparator},
	}
}

func generateLightningConfig() cloudv1.TomlConfig {
	lightningConfigTpl := `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`
	cfg := config.New(map[string]interface{}{})
	_ = cfg.UnmarshalTOML([]byte(lightningConfigTpl))
	return cloudv1.TomlConfig(*cfg)
}

func TestLightningConfig(t *testing.T) {
	var (
		fakeURL    = "fakeURL"
		fakeTCName = "fakeName"
	)
	cases := map[string]struct {
		source     cloudv1.Source
		modifyCfg  func(cloudv1.TomlConfig)
		targetPort int32
		delTLS     bool
		expectErr  error
		expectCfg  string
	}{
		fmt.Sprintf("ImportCSV_all_nil"): {
			source: generateCSVSource(nil, nil, nil, nil, nil, nil, nil),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("ImportCSV_sep"): {
			source: generateCSVSource(pointer.StringPtr("fakeSep"), nil, nil, nil, nil, nil, nil),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = "fakeSep"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("ImportCSV_delimiter"): {
			source: generateCSVSource(nil, pointer.StringPtr("fakeDel"), nil, nil, nil, nil, nil),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "fakeDel"
    header = true
    not-null = false
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("ImportCSV_null"): {
			source: generateCSVSource(nil, nil, pointer.StringPtr("fakeNull"), nil, nil, nil, nil),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "fakeNull"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("ImportCSV_header"): {
			source: generateCSVSource(nil, nil, nil, pointer.BoolPtr(false), nil, nil, nil),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = false
    not-null = false
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("ImportCSV_not-null"): {
			source: generateCSVSource(nil, nil, nil, nil, pointer.BoolPtr(true), nil, nil),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = true
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("ImportCSV_backslash_escape"): {
			source: generateCSVSource(nil, nil, nil, nil, nil, pointer.BoolPtr(false), nil),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = false
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("ImportCSV_trim_last_sep"): {
			source: generateCSVSource(nil, nil, nil, nil, nil, nil, pointer.BoolPtr(true)),
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","
    trim-last-separator = true

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)(?:\\.([0-9]+))?\\.csv$"
    schema = "$1"
    table = "$2"
    type = "csv"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("Import_Data"): {
			source: cloudv1.Source{Format: cloudv1.DataFormatAuroraParquet},
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = true
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([a-z0-9_]+)\\.([a-z0-9_]+)/(?:[^/]*/)*(?:[a-z0-9\\-_.]+\\.(parquet))$"
    schema = "$1"
    table = "$2"
    type = "$3"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("Import_DelTLS"): {
			source: cloudv1.Source{Format: cloudv1.DataFormatAuroraParquet},
			delTLS: true,
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = true
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([a-z0-9_]+)\\.([a-z0-9_]+)/(?:[^/]*/)*(?:[a-z0-9\\-_.]+\\.(parquet))$"
    schema = "$1"
    table = "$2"
    type = "$3"

[post-restore]
  analyze = true
  checksum = true

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("Import_NoSchema"): {
			source: cloudv1.Source{Format: cloudv1.DataFormatUnknown},
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = false
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 4000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
		fmt.Sprintf("Import_SpecificPort"): {
			source: cloudv1.Source{Format: cloudv1.DataFormatAuroraParquet},
			modifyCfg: func(cfg cloudv1.TomlConfig) {
				config.New(cfg.MP).Del("tidb.port")
			},
			targetPort: 5000,
			expectCfg: `[checkpoint]
  driver = "file"
  dsn = "/etc/endpoint/cp.pb"
  enable = true

[cron]
  log-progress = "5m"
  switch-mode = "5m"

[lightning]
  check-requirements = false
  file = ""
  index-concurrency = 4
  level = "info"
  pprof-port = 8289
  table-concurrency = 6

[mydumper]
  data-source-dir = "fakeURL"
  no-schema = true
  strict-format = false
  [mydumper.csv]
    backslash-escape = true
    delimiter = "\""
    header = true
    not-null = false
    null = "\\N"
    separator = ","

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)-schema-create\\.sql$"
    schema = "$1"
    type = "schema-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([^/.]+)\\.(.*?)-schema\\.sql$"
    schema = "$1"
    table = "$2"
    type = "table-schema"

  [[mydumper.files]]
    pattern = "(?i)^(?:[^/]*/)*([a-z0-9_]+)\\.([a-z0-9_]+)/(?:[^/]*/)*(?:[a-z0-9\\-_.]+\\.(parquet))$"
    schema = "$1"
    table = "$2"
    type = "$3"

[post-restore]
  analyze = true
  checksum = true

[security]
  ca-path = "/var/lib/lightning-tls/ca.crt"
  cert-path = "/var/lib/lightning-tls/tls.crt"
  key-path = "/var/lib/lightning-tls/tls.key"

[tidb]
  build-stats-concurrency = 4
  distsql-scan-concurrency = 15
  host = "fakeName-tidb"
  index-serial-scan-concurrency = 2
  log-level = "error"
  password = ""
  pd-addr = "fakeName-pd:2379"
  port = 5000
  status-port = 10080
  tls = "false"
  user = ""

[tikv-importer]
  backend = "local"
  sorted-kv-dir = "/etc/endpoint"
`,
		},
	}
	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			g := NewGomegaWithT(t)
			lcfg := generateLightningConfig()
			if tc.modifyCfg != nil {
				tc.modifyCfg(lcfg)
			}
			cfg, err := newLightningConfig(tc.source, lcfg, tc.delTLS, fakeURL, fakeTCName, tc.targetPort, "", "", "")
			if err != nil {
				g.Expect(err).To(Equal(tc.expectErr))
			} else {
				g.Expect(cfg).To(Equal(tc.expectCfg))
			}
		})
	}
}

func TestGenerateObjectNamePatternRegexp(t *testing.T) {
	wildcardsWorks := map[string][]string{
		`db1.tbl1`:                    {"db1.tbl1"},
		`db[0-9].tbl[0-9a-f][0-9a-f]`: {"db0.tblaa", "db0.tbla0"},
		`data.*`:                      {"data.blabla", "data.bla"},
		`*.backup_*`:                  {"hello.backup_2021"},
		`employees.*`:                 {"employees.1", "employees.1ajskldla"},
		`*.WorkOrder`:                 {"hello.WorkOrder"},
		`!*.Password`:                 {"!hello.Password"},
		`!employees.salaries`:         {"!employees.salaries"},
		`db.with.dots.*`:              {"db.with.dots.hello"},
		`"db.with.dots"."tbl\1"`:      {`"db.with.dots"."tbl\1"`},
		"`db.with.dots`.`tbl\\2`":     {"`db.with.dots`.`tbl\\2`"},
		`"foo""bar".` + "foo``bar``":  {`"foo""bar".` + "foo``bar``"},
		`foo\"bar.` + "foo\\`bar":     {`foo\"bar.` + "foo\\`bar"},
		`[0-9]`:                       {"0"},
		`[a-z]`:                       {"a"},
		`[a-z][0-9][a-z]`:             {"a0b"},
		`^%$`:                         {"^%$"},
		`^.$*`:                        {`^.$*`},
		`hello`:                       {`hello`},
		`hello.`:                      {`hello.`},
		`hello*`:                      {`hello.*`, `helloabcdksdk`},
		`hello?`:                      {`hello.`, `helloa`, `hello?`},
		`hello=world\\`:               {`hello=world\\`},
		`*hello=123?`:                 {`blabalhello=123a`},
		`*hello=123/*/abc?`:           {`*hello=123/ablablabl/abcd`},
		`*station=402260*`:            {`s3://wjktest123/mydata3/data/station=402260/part-r-00000-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`},
		`*station=402260*.parquet`:    {`s3://wjktest123/mydata3/data/station=402260/part-r-00000-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`},
		`*.parquet`:                   {`s3://wjktest123/mydata3/data/station=402265/part-r-00283-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`, `s3://wjktest123/mydata3/data/station=402260/part-r-00001-ddaee723-f3f6-4f25-a34b-3312172aa6d7.snappy.parquet`},
	}

	g := NewGomegaWithT(t)
	for k, vs := range wildcardsWorks {
		for _, v := range vs {
			generatedRegexp := generateObjectNamePatternRegexp(k)
			r, err := regexp.Compile(generatedRegexp)
			g.Expect(err).To(BeNil())
			g.Expect(r.MatchString(v)).To(BeTrue())
		}
	}

	wildcardsNotWorks := map[string][]string{
		"hello":      {"helloa", "hellob", "chello"},
		"hello[0-9]": {"helloa", "bhello"},
		"*hello":     {"ahellob", "hello*"},
		"hello?":     {"helloa.", "bhello"},
		"^%$":        {"^%$ww"},
		"^.$*":       {"?^*$*"},
	}
	for k, vs := range wildcardsNotWorks {
		for _, v := range vs {
			generatedRegexp := generateObjectNamePatternRegexp(k)
			r, err := regexp.Compile(generatedRegexp)
			g.Expect(err).To(BeNil())
			g.Expect(r.MatchString(v)).To(BeFalse())
		}
	}
}

func TestGenerateSingleTableSchemaFileRoute(t *testing.T) {
	type testCase struct {
		TargetTablePattern  string
		InputFiles          []string
		TableSchemaFileName string
		DBSchemaFileName    string
	}

	for _, tc := range []testCase{
		{
			TargetTablePattern: "db01.tbl01",
			InputFiles: []string{
				"s3://aaa/bbb/db01.tbl01-schema.sql",
				"db01-schema-create.sql",
				"aaa.csv",
			},
			TableSchemaFileName: "s3://aaa/bbb/db01.tbl01-schema.sql",
			DBSchemaFileName:    "db01-schema-create.sql",
		},
		{
			TargetTablePattern: "db_test01.tbl_test01",
			InputFiles: []string{
				"s3://aaa/bbb/db_test01.tbl_test01-schema.sql",
				"db_test01-schema-create.sql",
				"db_test01.sql",
				"db_test01.tbl_test01.sql",
				"aaa.csv",
			},
			TableSchemaFileName: "s3://aaa/bbb/db_test01.tbl_test01-schema.sql",
			DBSchemaFileName:    "db_test01-schema-create.sql",
		},
		{
			TargetTablePattern: "DB_TEST01.TBL_TEST01",
			InputFiles: []string{
				"s3://aaa/bbb/db_test01.tbl_test01-schema.sql",
				"s3://aaa/bbb/db_test01-schema-create.sql",
				"s3://aaa/bbb/DB_TEST01.sql",
				"s3://aaa/bbb/DB_TEST01.TBL_TEST01.sql",
				"aaa.csv",
			},
			TableSchemaFileName: "s3://aaa/bbb/db_test01.tbl_test01-schema.sql",
			DBSchemaFileName:    "s3://aaa/bbb/db_test01-schema-create.sql",
		},
		{
			TargetTablePattern: "db_test02.tbl_test02",
			InputFiles: []string{
				"s3://aaa/bbb/db_test01.tbl_test01-schema.sql",
				"s3://aaa/bbb/db_test02.tbl_test02-schema.sql",
				"s3://aaa/bbb/DB_TEST01.sql",
				"s3://aaa/bbb/DB_TEST01.TBL_TEST01.sql",
				"aaa.csv",
			},
			TableSchemaFileName: "s3://aaa/bbb/db_test02.tbl_test02-schema.sql",
			DBSchemaFileName:    "",
		},
		{
			TargetTablePattern: "db_test03.tbl_test03",
			InputFiles: []string{
				"db_test01.tbl_test03-schema.sql",
				"db_test03.tbl_test02-schema.sql",
				"DB_TEST01.sql",
				"DB_TEST01.TBL_TEST01.sql",
				"aaa.csv",
			},
			TableSchemaFileName: "",
			DBSchemaFileName:    "",
		},
	} {
		var (
			ok      bool
			matched bool
			err     error
			typeStr string
		)
		dbSchemaFileRoute, tblSchemaFileRoute, err := generateSingleTableSchemaFileRoute(tc.TargetTablePattern)
		require.NoError(t, err)
		t.Logf("%v, %v\n", dbSchemaFileRoute, tblSchemaFileRoute)

		typeStr, ok = dbSchemaFileRoute["type"]
		require.True(t, ok)
		require.Equal(t, "schema-schema", typeStr)

		dbRegExpStr, ok := dbSchemaFileRoute["pattern"]
		require.True(t, ok)

		typeStr, ok = tblSchemaFileRoute["type"]
		require.True(t, ok)
		require.Equal(t, "table-schema", typeStr)

		tblRegExpStr, ok := tblSchemaFileRoute["pattern"]
		require.True(t, ok)

		dbSchemaFileName := ""
		tblSchemaFileName := ""

		for _, f := range tc.InputFiles {
			matched, err = regexp.MatchString(dbRegExpStr, f)
			require.NoError(t, err)
			if matched {
				require.Equal(t, "", dbSchemaFileName) // ensure the db schema only match once
				dbSchemaFileName = f
			}
			matched, err = regexp.MatchString(tblRegExpStr, f)
			require.NoError(t, err)
			if matched {
				require.Equal(t, "", tblSchemaFileName) // ensure the table schema only match once
				tblSchemaFileName = f
			}
		}
		require.Equal(t, tc.DBSchemaFileName, dbSchemaFileName)
		require.Equal(t, tc.TableSchemaFileName, tblSchemaFileName)
	}
}