// Copyright 2023 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 cloudstorage

import (
	"fmt"
	"testing"

	"github.com/pingcap/ticdc/pkg/common"
	"github.com/stretchr/testify/require"
)

func TestSchemaPathKey(t *testing.T) {
	t.Parallel()

	testCases := []struct {
		path      string
		schemakey SchemaPathKey
		checksum  uint32
	}{
		// Test for database schema path: <schema>/meta/schema_{tableVersion}_{checksum}.json
		{
			path: "test_schema/meta/schema_1_2.json",
			schemakey: SchemaPathKey{
				Schema:       "test_schema",
				Table:        "",
				TableVersion: 1,
			},
			checksum: 2,
		},
		// Test for table schema path: <schema>/<table>/meta/schema_{tableVersion}_{checksum}.json
		{
			path: "test_schema/test_table/meta/schema_11_22.json",
			schemakey: SchemaPathKey{
				Schema:       "test_schema",
				Table:        "test_table",
				TableVersion: 11,
			},
			checksum: 22,
		},
	}
	for _, tc := range testCases {
		var schemaKey SchemaPathKey
		checksum, err := schemaKey.ParseSchemaFilePath(tc.path)
		require.NoError(t, err)
		require.Equal(t, tc.schemakey, schemaKey)
		require.Equal(t, tc.checksum, checksum)
	}
}

func TestIndexFileKey(t *testing.T) {
	t.Parallel()

	dispatcherID := common.NewDispatcherID()
	testCases := []struct {
		index          uint64
		fileIndexWidth int
		extension      string
		path           string
		indexPath      string
		dmlkey         DmlPathKey
	}{
		{
			index:          10,
			fileIndexWidth: 20,
			extension:      ".csv",
			path:           fmt.Sprintf("schema1/table1/123456/2023-05-09/CDC_%s_00000000000000000010.csv", dispatcherID.String()),
			indexPath:      fmt.Sprintf("schema1/table1/123456/2023-05-09/meta/CDC_%s.index", dispatcherID.String()),
			dmlkey: DmlPathKey{
				SchemaPathKey: SchemaPathKey{
					Schema:       "schema1",
					Table:        "table1",
					TableVersion: 123456,
				},
				PartitionNum: 0,
				Date:         "2023-05-09",
			},
		},
	}

	for _, tc := range testCases {
		var dmlkey DmlPathKey
		id, err := dmlkey.ParseIndexFilePath("day", tc.indexPath)
		require.NoError(t, err)
		require.Equal(t, tc.dmlkey, dmlkey)
		require.Equal(t, id, dispatcherID.String())

		fileIndex := &FileIndex{
			FileIndexKey: FileIndexKey{
				DispatcherID:           id,
				EnableTableAcrossNodes: id != "",
			},
			Idx: tc.index,
		}
		fileName := dmlkey.GenerateDMLFilePath(fileIndex, tc.extension, tc.fileIndexWidth)
		require.Equal(t, tc.path, fileName)
	}
}

func TestDmlPathKey(t *testing.T) {
	t.Parallel()

	dispatcherID := common.NewDispatcherID()
	testCases := []struct {
		index          int
		fileIndexWidth int
		extension      string
		path           string
		dmlkey         DmlPathKey
	}{
		{
			index:          10,
			fileIndexWidth: 20,
			extension:      ".csv",
			path:           fmt.Sprintf("schema1/table1/123456/2023-05-09/CDC_%s_00000000000000000010.csv", dispatcherID.String()),
			dmlkey: DmlPathKey{
				SchemaPathKey: SchemaPathKey{
					Schema:       "schema1",
					Table:        "table1",
					TableVersion: 123456,
				},
				PartitionNum: 0,
				Date:         "2023-05-09",
			},
		},
	}

	for _, tc := range testCases {
		var dmlkey DmlPathKey
		fileIndex, err := dmlkey.ParseDMLFilePath("day", tc.path)
		require.NoError(t, err)
		require.Equal(t, tc.dmlkey, dmlkey)

		fileName := dmlkey.GenerateDMLFilePath(fileIndex, tc.extension, tc.fileIndexWidth)
		require.Equal(t, tc.path, fileName)
	}
}