package engine

import (
	"context"
	"errors"
	"strings"
	"testing"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

func TestAzureBlobNamesContainRealObject(t *testing.T) {
	const prefix = "deltalake/org/cluster/persisted_statements_summary/_delta_log/"
	tests := []struct {
		name  string
		names []string
		want  bool
	}{
		{name: "empty listing", want: false},
		{name: "prefix marker", names: []string{prefix}, want: false},
		{
			name: "nested directory markers",
			names: []string{
				prefix,
				prefix + "checkpoints/",
			},
			want: false,
		},
		{
			name:  "delta commit",
			names: []string{prefix, prefix + "00000000000000000000.json"},
			want:  true,
		},
		{
			name:  "last checkpoint",
			names: []string{prefix + "_last_checkpoint"},
			want:  true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := azureBlobNamesContainRealObject(prefix, tt.names); got != tt.want {
				t.Fatalf("azureBlobNamesContainRealObject() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestClassifyAzureDeltaAvailability(t *testing.T) {
	tests := []struct {
		name          string
		hasDeltaLog   bool
		hasRootObject bool
		want          bool
		wantErr       error
	}{
		{name: "delta log exists", hasDeltaLog: true, hasRootObject: true, want: true},
		{name: "empty root", want: false},
		{
			name:          "non delta objects",
			hasRootObject: true,
			wantErr:       ErrInvalidDeltaSource,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := classifyAzureDeltaAvailability("https://account/container/table", tt.hasDeltaLog, tt.hasRootObject)
			if got != tt.want {
				t.Fatalf("classifyAzureDeltaAvailability() = %v, want %v", got, tt.want)
			}
			if tt.wantErr == nil {
				if err != nil {
					t.Fatalf("classifyAzureDeltaAvailability() error = %v", err)
				}
				return
			}
			if !errors.Is(err, tt.wantErr) {
				t.Fatalf("classifyAzureDeltaAvailability() error = %v, want %v", err, tt.wantErr)
			}
		})
	}
}

func TestAzureDeltaBlobTarget(t *testing.T) {
	t.Run("canonical URL", func(t *testing.T) {
		info, err := dto.ParseDataSource("https://account.dfs.core.windows.net/container/deltalake/org/cluster/persisted_statements_summary/")
		if err != nil {
			t.Fatal(err)
		}

		serviceURL, container, prefix, err := azureDeltaBlobTarget(info)
		if err != nil {
			t.Fatalf("azureDeltaBlobTarget() error = %v", err)
		}
		if serviceURL != "https://account.blob.core.windows.net/" {
			t.Fatalf("serviceURL = %q", serviceURL)
		}
		if container != "container" {
			t.Fatalf("container = %q", container)
		}
		if prefix != "deltalake/org/cluster/persisted_statements_summary" {
			t.Fatalf("prefix = %q", prefix)
		}
	})

	t.Run("short ABFSS uses region naming", func(t *testing.T) {
		info, err := dto.ParseDataSource("abfss://o11y-prod-shared-japaneast/deltalake/org/cluster/persisted_statements_summary/")
		if err != nil {
			t.Fatal(err)
		}
		info.Region = "japaneast"

		serviceURL, container, prefix, err := azureDeltaBlobTarget(info)
		if err != nil {
			t.Fatalf("azureDeltaBlobTarget() error = %v", err)
		}
		if serviceURL != "https://o11yprodjapaneast.blob.core.windows.net/" {
			t.Fatalf("serviceURL = %q", serviceURL)
		}
		if container != "o11y-prod-shared-japaneast" {
			t.Fatalf("container = %q", container)
		}
		if prefix != "deltalake/org/cluster/persisted_statements_summary" {
			t.Fatalf("prefix = %q", prefix)
		}
	})

	t.Run("missing account and container use region naming", func(t *testing.T) {
		info := &dto.DataSourceInfo{
			Region: "japaneast",
			Path:   "deltalake/org/cluster/persisted_statements_summary",
		}

		serviceURL, container, prefix, err := azureDeltaBlobTarget(info)
		if err != nil {
			t.Fatalf("azureDeltaBlobTarget() error = %v", err)
		}
		if serviceURL != "https://o11yprodjapaneast.blob.core.windows.net/" {
			t.Fatalf("serviceURL = %q", serviceURL)
		}
		if container != "o11y-prod-shared-japaneast" {
			t.Fatalf("container = %q", container)
		}
		if prefix != "deltalake/org/cluster/persisted_statements_summary" {
			t.Fatalf("prefix = %q", prefix)
		}
	})
}

func TestSynapseHasDeltaLogRejectsSQLAuth(t *testing.T) {
	engine := &SynapseEngine{authType: "sql"}
	_, err := engine.HasDeltaLog(context.Background(), nil)
	if err == nil || !strings.Contains(err.Error(), "unsupported for Synapse SQL auth") {
		t.Fatalf("HasDeltaLog() error = %v, want SQL auth unsupported", err)
	}
}