// Copyright 2026 Google LLC
//
// 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,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package serverlessspark_test

import (
	"testing"
	"time"

	"cloud.google.com/go/dataproc/v2/apiv1/dataprocpb"
	"github.com/googleapis/mcp-toolbox/internal/sources/serverlessspark"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func TestExtractBatchDetails_Success(t *testing.T) {
	batchName := "projects/my-project/locations/us-central1/batches/my-batch"
	projectID, location, batchID, err := serverlessspark.ExtractBatchDetails(batchName)
	if err != nil {
		t.Errorf("ExtractBatchDetails() error = %v, want no error", err)
		return
	}
	wantProject := "my-project"
	wantLocation := "us-central1"
	wantBatchID := "my-batch"
	if projectID != wantProject {
		t.Errorf("ExtractBatchDetails() projectID = %v, want %v", projectID, wantProject)
	}
	if location != wantLocation {
		t.Errorf("ExtractBatchDetails() location = %v, want %v", location, wantLocation)
	}
	if batchID != wantBatchID {
		t.Errorf("ExtractBatchDetails() batchID = %v, want %v", batchID, wantBatchID)
	}
}

func TestExtractBatchDetails_Failure(t *testing.T) {
	batchName := "invalid-name"
	_, _, _, err := serverlessspark.ExtractBatchDetails(batchName)
	wantErr := "failed to parse batch name: invalid-name"
	if err == nil || err.Error() != wantErr {
		t.Errorf("ExtractBatchDetails() error = %v, want %v", err, wantErr)
	}
}

func TestBatchConsoleURL(t *testing.T) {
	got := serverlessspark.BatchConsoleURL("my-project", "us-central1", "my-batch")
	want := "https://console.cloud.google.com/dataproc/batches/us-central1/my-batch/summary?project=my-project"
	if got != want {
		t.Errorf("BatchConsoleURL() = %v, want %v", got, want)
	}
}

func TestBatchLogsURL(t *testing.T) {
	startTime := time.Date(2025, 10, 1, 5, 0, 0, 0, time.UTC)
	endTime := time.Date(2025, 10, 1, 6, 0, 0, 0, time.UTC)
	got := serverlessspark.BatchLogsURL("my-project", "us-central1", "my-batch", startTime, endTime)
	want := "https://console.cloud.google.com/logs/viewer?advancedFilter=" +
		"resource.type%3D%22cloud_dataproc_batch%22" +
		"%0Aresource.labels.project_id%3D%22my-project%22" +
		"%0Aresource.labels.location%3D%22us-central1%22" +
		"%0Aresource.labels.batch_id%3D%22my-batch%22" +
		"%0Atimestamp%3E%3D%222025-10-01T04%3A59%3A00Z%22" + // Minus 1 minute
		"%0Atimestamp%3C%3D%222025-10-01T06%3A10%3A00Z%22" + // Plus 10 minutes
		"&project=my-project" +
		"&resource=cloud_dataproc_batch%2Fbatch_id%2Fmy-batch"
	if got != want {
		t.Errorf("BatchLogsURL() = %v, want %v", got, want)
	}
}

func TestBatchConsoleURLFromProto(t *testing.T) {
	batchPb := &dataprocpb.Batch{
		Name: "projects/my-project/locations/us-central1/batches/my-batch",
	}
	got, err := serverlessspark.BatchConsoleURLFromProto(batchPb)
	if err != nil {
		t.Fatalf("BatchConsoleURLFromProto() error = %v", err)
	}
	want := "https://console.cloud.google.com/dataproc/batches/us-central1/my-batch/summary?project=my-project"
	if got != want {
		t.Errorf("BatchConsoleURLFromProto() = %v, want %v", got, want)
	}
}

func TestBatchLogsURLFromProto(t *testing.T) {
	createTime := time.Date(2025, 10, 1, 5, 0, 0, 0, time.UTC)
	stateTime := time.Date(2025, 10, 1, 6, 0, 0, 0, time.UTC)
	batchPb := &dataprocpb.Batch{
		Name:       "projects/my-project/locations/us-central1/batches/my-batch",
		CreateTime: timestamppb.New(createTime),
		StateTime:  timestamppb.New(stateTime),
	}
	got, err := serverlessspark.BatchLogsURLFromProto(batchPb)
	if err != nil {
		t.Fatalf("BatchLogsURLFromProto() error = %v", err)
	}
	want := "https://console.cloud.google.com/logs/viewer?advancedFilter=" +
		"resource.type%3D%22cloud_dataproc_batch%22" +
		"%0Aresource.labels.project_id%3D%22my-project%22" +
		"%0Aresource.labels.location%3D%22us-central1%22" +
		"%0Aresource.labels.batch_id%3D%22my-batch%22" +
		"%0Atimestamp%3E%3D%222025-10-01T04%3A59%3A00Z%22" + // Minus 1 minute
		"%0Atimestamp%3C%3D%222025-10-01T06%3A10%3A00Z%22" + // Plus 10 minutes
		"&project=my-project" +
		"&resource=cloud_dataproc_batch%2Fbatch_id%2Fmy-batch"
	if got != want {
		t.Errorf("BatchLogsURLFromProto() = %v, want %v", got, want)
	}
}

func TestExtractSessionDetails_Success(t *testing.T) {
	sessionName := "projects/my-project/locations/us-central1/sessions/my-session"
	projectID, location, sessionID, err := serverlessspark.ExtractSessionDetails(sessionName)
	if err != nil {
		t.Errorf("ExtractSessionDetails() error = %v, want no error", err)
		return
	}
	wantProject := "my-project"
	wantLocation := "us-central1"
	wantSessionID := "my-session"
	if projectID != wantProject {
		t.Errorf("ExtractSessionDetails() projectID = %v, want %v", projectID, wantProject)
	}
	if location != wantLocation {
		t.Errorf("ExtractSessionDetails() location = %v, want %v", location, wantLocation)
	}
	if sessionID != wantSessionID {
		t.Errorf("ExtractSessionDetails() sessionID = %v, want %v", sessionID, wantSessionID)
	}
}

func TestExtractSessionDetails_Failure(t *testing.T) {
	sessionName := "invalid-name"
	_, _, _, err := serverlessspark.ExtractSessionDetails(sessionName)
	wantErr := "failed to parse session name: invalid-name"
	if err == nil || err.Error() != wantErr {
		t.Errorf("ExtractSessionDetails() error = %v, want %v", err, wantErr)
	}
}

func TestSessionConsoleURL(t *testing.T) {
	got := serverlessspark.SessionConsoleURL("my-project", "us-central1", "my-session")
	want := "https://console.cloud.google.com/dataproc/interactive/us-central1/my-session/details?project=my-project"
	if got != want {
		t.Errorf("SessionConsoleURL() = %v, want %v", got, want)
	}
}

func TestSessionLogsURL(t *testing.T) {
	startTime := time.Date(2025, 10, 1, 5, 0, 0, 0, time.UTC)
	endTime := time.Date(2025, 10, 1, 6, 0, 0, 0, time.UTC)
	got := serverlessspark.SessionLogsURL("my-project", "us-central1", "my-session", startTime, endTime)
	want := "https://console.cloud.google.com/logs/viewer?advancedFilter=" +
		"resource.type%3D%22cloud_dataproc_session%22" +
		"%0Aresource.labels.session_id%3D%22my-session%22" +
		"%0Aresource.labels.project_id%3D%22my-project%22" +
		"%0Aresource.labels.location%3D%22us-central1%22" +
		"%0Atimestamp%3E%3D%222025-10-01T04%3A59%3A00Z%22" + // Minus 1 minute
		"%0Atimestamp%3C%3D%222025-10-01T06%3A10%3A00Z%22" + // Plus 10 minutes
		"&project=my-project"
	if got != want {
		t.Errorf("SessionLogsURL() = \n%v\nwant \n%v", got, want)
	}
}

func TestSessionConsoleURLFromProto(t *testing.T) {
	sessionPb := &dataprocpb.Session{
		Name: "projects/my-project/locations/us-central1/sessions/my-session",
	}
	got, err := serverlessspark.SessionConsoleURLFromProto(sessionPb)
	if err != nil {
		t.Fatalf("SessionConsoleURLFromProto() error = %v", err)
	}
	want := "https://console.cloud.google.com/dataproc/interactive/us-central1/my-session/details?project=my-project"
	if got != want {
		t.Errorf("SessionConsoleURLFromProto() = %v, want %v", got, want)
	}
}

func TestSessionLogsURLFromProto(t *testing.T) {
	createTime := time.Date(2025, 10, 1, 5, 0, 0, 0, time.UTC)
	stateTime := time.Date(2025, 10, 1, 6, 0, 0, 0, time.UTC)
	sessionPb := &dataprocpb.Session{
		Name:       "projects/my-project/locations/us-central1/sessions/my-session",
		CreateTime: timestamppb.New(createTime),
		StateTime:  timestamppb.New(stateTime),
	}
	got, err := serverlessspark.SessionLogsURLFromProto(sessionPb)
	if err != nil {
		t.Fatalf("SessionLogsURLFromProto() error = %v", err)
	}
	want := "https://console.cloud.google.com/logs/viewer?advancedFilter=" +
		"resource.type%3D%22cloud_dataproc_session%22" +
		"%0Aresource.labels.session_id%3D%22my-session%22" +
		"%0Aresource.labels.project_id%3D%22my-project%22" +
		"%0Aresource.labels.location%3D%22us-central1%22" +
		"%0Atimestamp%3E%3D%222025-10-01T04%3A59%3A00Z%22" + // Minus 1 minute
		"%0Atimestamp%3C%3D%222025-10-01T06%3A10%3A00Z%22" + // Plus 10 minutes
		"&project=my-project"
	if got != want {
		t.Errorf("SessionLogsURLFromProto() = \n%v\nwant \n%v", got, want)
	}
}

func TestSessionLogsURL_Escaping(t *testing.T) {
	startTime := time.Date(2025, 10, 1, 5, 0, 0, 0, time.UTC)
	endTime := time.Date(2025, 10, 1, 6, 0, 0, 0, time.UTC)

	// Input contains a double quote which should be escaped.
	sessionID := `my-session" OR root`
	got := serverlessspark.SessionLogsURL("my-project", "us-central1", sessionID, startTime, endTime)

	want := "https://console.cloud.google.com/logs/viewer?advancedFilter=" +
		"resource.type%3D%22cloud_dataproc_session%22" +
		// "my-session\" OR root" encoded
		"%0Aresource.labels.session_id%3D%22my-session%5C%22+OR+root%22" +
		"%0Aresource.labels.project_id%3D%22my-project%22" +
		"%0Aresource.labels.location%3D%22us-central1%22" +
		"%0Atimestamp%3E%3D%222025-10-01T04%3A59%3A00Z%22" +
		"%0Atimestamp%3C%3D%222025-10-01T06%3A10%3A00Z%22" +
		"&project=my-project"

	if got != want {
		t.Errorf("SessionLogsURL_Escaping() = \n%v\nwant \n%v", got, want)
	}
}

func TestExtractSessionTemplateDetails_Success(t *testing.T) {
	sessionTemplateName := "projects/my-project/locations/us-central1/sessionTemplates/my-session-template"
	projectID, location, sessionTemplateID, err := serverlessspark.ExtractSessionTemplateDetails(sessionTemplateName)
	if err != nil {
		t.Errorf("ExtractSessionTemplateDetails() error = %v, want no error", err)
		return
	}
	wantProject := "my-project"
	wantLocation := "us-central1"
	wantSessionTemplateID := "my-session-template"
	if projectID != wantProject {
		t.Errorf("ExtractSessionTemplateDetails() projectID = %v, want %v", projectID, wantProject)
	}
	if location != wantLocation {
		t.Errorf("ExtractSessionTemplateDetails() location = %v, want %v", location, wantLocation)
	}
	if sessionTemplateID != wantSessionTemplateID {
		t.Errorf("ExtractSessionTemplateDetails() sessionTemplateID = %v, want %v", sessionTemplateID, wantSessionTemplateID)
	}
}

func TestExtractSessionTemplateDetails_Failure(t *testing.T) {
	sessionTemplateName := "invalid-name"
	_, _, _, err := serverlessspark.ExtractSessionTemplateDetails(sessionTemplateName)
	wantErr := "failed to parse session template name: invalid-name"
	if err == nil || err.Error() != wantErr {
		t.Errorf("ExtractSessionTemplateDetails() error = %v, want %v", err, wantErr)
	}
}