/*
Copyright 2024.

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 integrationtask

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"

	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/iam"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

type getErrorClient struct {
	client.Client
	err error
}

func (c *getErrorClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
	return c.err
}

func setupTestTask() *IntegrationTask {
	serviceInfo := structs.ServiceInfo{
		Vendor: "aws",
		Region: "us-west-2",
		AccountInfo: structs.AccountInfoConfig{
			Account: "123456789012",
			Oidc:    "oidc.eks.us-west-2.amazonaws.com/id/EXAMPLE",
		},
		BizTypeConfigs: map[string]structs.BizTypeConfig{
			"tidbcloud": {
				ExternalBucket: "tidb-shared-data",
			},
		},
	}

	return &IntegrationTask{
		serviceInfo: serviceInfo,
	}
}

func setupTestTaskAliCloud() *IntegrationTask {
	serviceInfo := structs.ServiceInfo{
		Vendor: "alicloud",
		Region: "cn-hangzhou",
		AccountInfo: structs.AccountInfoConfig{
			Account: "123456789012",
			Oidc:    "oidc-provider-example",
		},
		BizTypeConfigs: map[string]structs.BizTypeConfig{
			"tidbcloud": {
				ExternalBucket: "tidb-shared-data",
			},
		},
	}

	return &IntegrationTask{
		serviceInfo: serviceInfo,
	}
}

func TestIntegrationTask_ExtractTenantInfo_Success(t *testing.T) {
	task := setupTestTask()

	integrationTask := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				"observability.pingcap.com/tenantID": "tenant123",
				"observability.pingcap.com/bizType":  "tidbcloud",
			},
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			TaskType:      observabilitypingcapcomv1.SlowQueryTask,
			ClusterID:     "test-cluster",
			AccessRoleARN: "arn:aws:iam::123456789012:role/test-role",
			Frequency:     "10m",
		},
	}

	tenantID, bizType := task.extractTenantInfo(integrationTask)
	if tenantID != "tenant123" {
		t.Errorf("Expected tenant ID 'tenant123', got '%s'", tenantID)
	}
	if bizType != "tidbcloud" {
		t.Errorf("Expected biz type 'tidbcloud', got '%s'", bizType)
	}
}

func TestIntegrationTask_ExtractTenantInfo_NoLabels(t *testing.T) {
	task := setupTestTask()

	integrationTask := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			TaskType:      observabilitypingcapcomv1.SlowQueryTask,
			ClusterID:     "test-cluster",
			AccessRoleARN: "arn:aws:iam::123456789012:role/test-role",
		},
	}

	tenantID, bizType := task.extractTenantInfo(integrationTask)
	if tenantID != "" {
		t.Errorf("Expected empty tenant ID, got '%s'", tenantID)
	}
	if bizType != "" {
		t.Errorf("Expected empty biz type, got '%s'", bizType)
	}
}

func TestIntegrationTask_ExtractTenantInfo_PartialLabels(t *testing.T) {
	task := setupTestTask()

	integrationTask := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				"observability.pingcap.com/tenantID": "tenant789",
				// Missing bizType label
			},
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			TaskType:      observabilitypingcapcomv1.SlowQueryTask,
			ClusterID:     "test-cluster",
			AccessRoleARN: "arn:aws:iam::123456789012:role/test-role",
		},
	}

	tenantID, bizType := task.extractTenantInfo(integrationTask)
	if tenantID != "tenant789" {
		t.Errorf("Expected tenant ID 'tenant789', got '%s'", tenantID)
	}
	if bizType != "" {
		t.Errorf("Expected empty biz type, got '%s'", bizType)
	}
}

func TestIntegrationTask_ServiceInfoVendor_AWS(t *testing.T) {
	task := &IntegrationTask{
		serviceInfo: structs.ServiceInfo{
			Vendor: "aws",
			Region: "us-west-2",
		},
	}

	if task.serviceInfo.Vendor != "aws" {
		t.Errorf("Expected vendor 'aws', got '%s'", task.serviceInfo.Vendor)
	}

	if task.serviceInfo.Region != "us-west-2" {
		t.Errorf("Expected region 'us-west-2', got '%s'", task.serviceInfo.Region)
	}
}

func TestIntegrationTask_ServiceInfoVendor_AliCloud(t *testing.T) {
	task := &IntegrationTask{
		serviceInfo: structs.ServiceInfo{
			Vendor: "alicloud",
			Region: "cn-hangzhou",
		},
	}

	if task.serviceInfo.Vendor != "alicloud" {
		t.Errorf("Expected vendor 'alicloud', got '%s'", task.serviceInfo.Vendor)
	}

	if task.serviceInfo.Region != "cn-hangzhou" {
		t.Errorf("Expected region 'cn-hangzhou', got '%s'", task.serviceInfo.Region)
	}
}

func TestIntegrationTask_ServiceInfoVendor_Unknown(t *testing.T) {
	task := &IntegrationTask{
		serviceInfo: structs.ServiceInfo{
			Vendor: "unknown",
			Region: "us-west-2",
		},
	}

	if task.serviceInfo.Vendor != "unknown" {
		t.Errorf("Expected vendor 'unknown', got '%s'", task.serviceInfo.Vendor)
	}

	if task.serviceInfo.Region != "us-west-2" {
		t.Errorf("Expected region 'us-west-2', got '%s'", task.serviceInfo.Region)
	}
}

func TestIntegrationTask_CreateInitialAssumeRolePolicy_AWS(t *testing.T) {
	task := setupTestTask()

	policy := task.createInitialAssumeRolePolicy("arn:aws:iam::123456789012:role/test-role", "test-namespace")

	// Check basic structure
	if policy.Version != "2012-10-17" {
		t.Errorf("Expected version '2012-10-17', got %v", policy.Version)
	}

	// Check that Statement exists and has at least 1 statement
	if len(policy.Statement) < 1 {
		t.Errorf("Expected at least 1 statement, got %d", len(policy.Statement))
	}

	// Convert to JSON to check content
	policyJSON, err := json.Marshal(policy)
	if err != nil {
		t.Fatalf("Failed to marshal policy: %v", err)
	}

	policyStr := string(policyJSON)
	if !strings.Contains(policyStr, "arn:aws:iam::123456789012:role/test-role") {
		t.Errorf("Expected policy to contain role ARN, got %s", policyStr)
	}

	if !strings.Contains(policyStr, "sts:AssumeRole") {
		t.Errorf("Expected policy to contain 'sts:AssumeRole' action, got %s", policyStr)
	}
}

func TestIntegrationTask_CreateInitialAssumeRolePolicy_AliCloud(t *testing.T) {
	task := setupTestTaskAliCloud()

	policy := task.createInitialAssumeRolePolicy("acs:ram::123456789012:role/test-role", "test-namespace")

	// Check basic structure
	if policy.Version != "1" {
		t.Errorf("Expected version '1', got %v", policy.Version)
	}

	// Check that Statement exists and has at least 1 statement
	if len(policy.Statement) < 1 {
		t.Errorf("Expected at least 1 statement, got %d", len(policy.Statement))
	}

	// Convert to JSON to check content
	policyJSON, err := json.Marshal(policy)
	if err != nil {
		t.Fatalf("Failed to marshal policy: %v", err)
	}

	policyStr := string(policyJSON)
	if !strings.Contains(policyStr, "acs:ram::123456789012:role/test-role") {
		t.Errorf("Expected policy to contain role ARN, got %s", policyStr)
	}

	if !strings.Contains(policyStr, "sts:AssumeRole") {
		t.Errorf("Expected policy to contain 'sts:AssumeRole' action, got %s", policyStr)
	}
}

func TestIntegrationTask_CreateTenantWideAccessPolicy_AWS(t *testing.T) {
	task := setupTestTask()

	policy := task.createTenantWideAccessPolicy("tenant123", "tidbcloud")

	// Check basic structure
	if policy["Version"] != "2012-10-17" {
		t.Errorf("Expected version '2012-10-17', got %v", policy["Version"])
	}

	// Check that Statement exists
	if _, ok := policy["Statement"]; !ok {
		t.Error("Expected 'Statement' field in policy")
	}

	// Convert to JSON to check content
	policyJSON, err := json.Marshal(policy)
	if err != nil {
		t.Fatalf("Failed to marshal policy: %v", err)
	}

	policyStr := string(policyJSON)

	// Should contain tenant-wide path pattern with wildcard for cluster
	if !strings.Contains(policyStr, "log_data/data/tenant123/*") {
		t.Errorf("Expected policy to contain tenant-wide path 'log_data/data/tenant123/*', got %s", policyStr)
	}

	// Should contain S3 actions
	if !strings.Contains(policyStr, "s3:GetObject") {
		t.Errorf("Expected policy to contain 's3:GetObject' action, got %s", policyStr)
	}
}

func TestIntegrationTask_CreateTenantWideAccessPolicy_AliCloud(t *testing.T) {
	task := setupTestTaskAliCloud()

	policy := task.createTenantWideAccessPolicy("tenant123", "tidbcloud")

	// Check basic structure
	if policy["Version"] != "1" {
		t.Errorf("Expected version '1', got %v", policy["Version"])
	}

	// Check that Statement exists
	if _, ok := policy["Statement"]; !ok {
		t.Error("Expected 'Statement' field in policy")
	}

	// Convert to JSON to check content
	policyJSON, err := json.Marshal(policy)
	if err != nil {
		t.Fatalf("Failed to marshal policy: %v", err)
	}

	policyStr := string(policyJSON)

	// Should contain tenant-wide path pattern
	if !strings.Contains(policyStr, "log_data/data/tenant123") {
		t.Errorf("Expected policy to contain tenant-wide path 'log_data/data/tenant123', got %s", policyStr)
	}

	// Should contain OSS actions
	if !strings.Contains(policyStr, "oss:GetObject") {
		t.Errorf("Expected policy to contain 'oss:GetObject' action, got %s", policyStr)
	}
}

func TestIntegrationTask_CheckRemainingPrincipals_AWS_HasPrincipals(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"AWS": []string{"arn:aws:iam::123456789012:role/role1", "arn:aws:iam::123456789012:role/role2"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	hasPrincipals := task.checkRemainingPrincipals(policyData, "AWS")
	if !hasPrincipals {
		t.Error("Expected to find remaining principals, but got false")
	}
}

func TestIntegrationTask_CheckRemainingPrincipals_AWS_NoPrincipals(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"Federated": "arn:aws:iam::123456789012:oidc-provider/example",
				},
				Action: "sts:AssumeRoleWithWebIdentity",
			},
		},
	}

	hasPrincipals := task.checkRemainingPrincipals(policyData, "AWS")
	if hasPrincipals {
		t.Error("Expected no remaining principals, but got true")
	}
}

func TestIntegrationTask_CheckRemainingPrincipals_AliCloud_HasPrincipals(t *testing.T) {
	task := setupTestTaskAliCloud()

	policyData := &iam.AssumeRolePolicy{
		Version: "1",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"RAM": []string{"acs:ram::123456789012:role/role1"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	hasPrincipals := task.checkRemainingPrincipals(policyData, "RAM")
	if !hasPrincipals {
		t.Error("Expected to find remaining principals, but got false")
	}
}

func TestIntegrationTask_CheckRemainingPrincipals_EmptyArray(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"AWS": []string{},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	hasPrincipals := task.checkRemainingPrincipals(policyData, "AWS")
	if hasPrincipals {
		t.Error("Expected no remaining principals with empty array, but got true")
	}
}

func TestIntegrationTask_RemovePrincipalFromPolicy_AWS_SinglePrincipal(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"AWS": []string{"arn:aws:iam::123456789012:role/test-role"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	removed, hasPrincipalsLeft, err := task.removePrincipalFromPolicy(policyData, "arn:aws:iam::123456789012:role/test-role")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if !removed {
		t.Error("Expected principal to be removed")
	}
	if hasPrincipalsLeft {
		t.Error("Expected no principals left after removing the only principal")
	}
}

func TestIntegrationTask_RemovePrincipalFromPolicy_AWS_MultiplePrincipals(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"AWS": []string{
						"arn:aws:iam::123456789012:role/role1",
						"arn:aws:iam::123456789012:role/role2",
					},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	removed, hasPrincipalsLeft, err := task.removePrincipalFromPolicy(policyData, "arn:aws:iam::123456789012:role/role1")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if !removed {
		t.Error("Expected principal to be removed")
	}
	if !hasPrincipalsLeft {
		t.Error("Expected principals to remain after removing one")
	}

	// Verify the remaining principal
	principals := policyData.Statement[0].Principal["AWS"].([]string)
	if len(principals) != 1 {
		t.Errorf("Expected 1 remaining principal, got %d", len(principals))
	}
	if principals[0] != "arn:aws:iam::123456789012:role/role2" {
		t.Errorf("Expected remaining principal 'role2', got '%s'", principals[0])
	}
}

func TestIntegrationTask_RemovePrincipalFromPolicy_AliCloud_SinglePrincipal(t *testing.T) {
	task := setupTestTaskAliCloud()

	policyData := &iam.AssumeRolePolicy{
		Version: "1",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"RAM": []string{"acs:ram::123456789012:role/test-role"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	removed, hasPrincipalsLeft, err := task.removePrincipalFromPolicy(policyData, "acs:ram::123456789012:role/test-role")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if !removed {
		t.Error("Expected principal to be removed")
	}
	if hasPrincipalsLeft {
		t.Error("Expected no principals left after removing the only principal")
	}
}

func TestIntegrationTask_RemovePrincipalFromPolicy_NotFound(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"AWS": []string{"arn:aws:iam::123456789012:role/role1"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	removed, hasPrincipalsLeft, err := task.removePrincipalFromPolicy(policyData, "arn:aws:iam::123456789012:role/nonexistent")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if removed {
		t.Error("Expected principal not to be removed when not found")
	}
	if !hasPrincipalsLeft {
		t.Error("Expected principals to remain when removing nonexistent principal")
	}
}

func TestIntegrationTask_AddPrincipalToPolicy_AWS_NewPrincipal(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"AWS": []string{"arn:aws:iam::123456789012:role/role1"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	updated, err := task.addPrincipalToPolicy(policyData, "arn:aws:iam::123456789012:role/role2")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if !updated {
		t.Error("Expected policy to be updated")
	}

	// Verify both principals exist
	principals := policyData.Statement[0].Principal["AWS"].([]string)
	if len(principals) != 2 {
		t.Errorf("Expected 2 principals, got %d", len(principals))
	}
}

func TestIntegrationTask_AddPrincipalToPolicy_AWS_DuplicatePrincipal(t *testing.T) {
	task := setupTestTask()

	policyData := &iam.AssumeRolePolicy{
		Version: "2012-10-17",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"AWS": []string{"arn:aws:iam::123456789012:role/role1"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	updated, err := task.addPrincipalToPolicy(policyData, "arn:aws:iam::123456789012:role/role1")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if updated {
		t.Error("Expected policy not to be updated when adding duplicate principal")
	}

	// Verify only one principal exists
	principals := policyData.Statement[0].Principal["AWS"].([]string)
	if len(principals) != 1 {
		t.Errorf("Expected 1 principal, got %d", len(principals))
	}
}

func TestIntegrationTask_AddPrincipalToPolicy_AliCloud_NewPrincipal(t *testing.T) {
	task := setupTestTaskAliCloud()

	policyData := &iam.AssumeRolePolicy{
		Version: "1",
		Statement: []iam.Statement{
			{
				Effect: "Allow",
				Principal: map[string]any{
					"RAM": []string{"acs:ram::123456789012:role/role1"},
				},
				Action: "sts:AssumeRole",
			},
		},
	}

	updated, err := task.addPrincipalToPolicy(policyData, "acs:ram::123456789012:role/role2")
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if !updated {
		t.Error("Expected policy to be updated")
	}

	// Verify both principals exist
	principals := policyData.Statement[0].Principal["RAM"].([]string)
	if len(principals) != 2 {
		t.Errorf("Expected 2 principals, got %d", len(principals))
	}
}

func TestIntegrationTask_BuildS3Path_AWS(t *testing.T) {
	task := setupTestTask()

	integrationTask := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				"observability.pingcap.com/tenantID":  "tenant123",
				"observability.pingcap.com/projectID": "project456",
			},
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			ClusterID:             "cluster789",
			DestinationBucketName: "test-bucket",
		},
	}

	path := task.buildS3Path(integrationTask)
	expected := "s3://test-bucket/log_data/data/tenant123/project456/cluster789"
	if path != expected {
		t.Errorf("Expected path '%s', got '%s'", expected, path)
	}
}

func TestIntegrationTask_BuildS3Path_AliCloud(t *testing.T) {
	task := setupTestTaskAliCloud()

	integrationTask := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				"observability.pingcap.com/tenantID":  "tenant123",
				"observability.pingcap.com/projectID": "project456",
			},
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			ClusterID:             "cluster789",
			DestinationBucketName: "test-bucket",
		},
	}

	path := task.buildS3Path(integrationTask)
	expected := "oss://test-bucket/log_data/data/tenant123/project456/cluster789"
	if path != expected {
		t.Errorf("Expected path '%s', got '%s'", expected, path)
	}
}

func setupIntegrationTaskWithFakeClient(t *testing.T, objs ...client.Object) *IntegrationTask {
	t.Helper()

	scheme := runtime.NewScheme()
	if err := observabilitypingcapcomv1.AddToScheme(scheme); err != nil {
		t.Fatalf("failed to add observability scheme: %v", err)
	}
	if err := corev1.AddToScheme(scheme); err != nil {
		t.Fatalf("failed to add corev1 scheme: %v", err)
	}

	fakeClient := fake.NewClientBuilder().
		WithScheme(scheme).
		WithObjects(objs...).
		Build()

	task := setupTestTask()
	task.cli = fakeClient
	return task
}

func TestIntegrationTask_CreateOrUpdateJobManagerConfig_CreateNew(t *testing.T) {
	it := setupIntegrationTaskWithFakeClient(t)

	task := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "itask-1",
			Namespace: "ns-test",
			Labels: map[string]string{
				constants.LabelProjectID: "project-1",
			},
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			ClusterID:  "cluster-1",
			Frequency:  "10m",
			TaskType:   observabilitypingcapcomv1.SlowQueryTask,
			DataFormat: "json",
		},
	}

	err := it.createOrUpdateJobManagerConfig(context.Background(), task, "ns-test", "tenant-1")
	if err != nil {
		t.Fatalf("createOrUpdateJobManagerConfig returned error: %v", err)
	}

	configMap := &corev1.ConfigMap{}
	err = it.cli.Get(context.Background(), client.ObjectKey{Name: "job-manager-config-tenant-1", Namespace: "ns-test"}, configMap)
	if err != nil {
		t.Fatalf("failed to get ConfigMap: %v", err)
	}

	if configMap.Data[configFileName] == "" {
		t.Fatalf("expected %s to be populated", configFileName)
	}

	var configData map[string]map[string]map[string]map[string]interface{}
	err = json.Unmarshal([]byte(configMap.Data[configFileName]), &configData)
	if err != nil {
		t.Fatalf("failed to parse config json: %v", err)
	}

	clusterCfg := configData["tenants"]["tenant-1"]["cluster-1"]
	if clusterCfg == nil {
		t.Fatalf("expected cluster configuration to exist")
	}
	if clusterCfg["frequency"] != "10m" {
		t.Fatalf("unexpected frequency: %v", clusterCfg["frequency"])
	}
	if clusterCfg["taskType"] != "slowquery" {
		t.Fatalf("unexpected taskType: %v", clusterCfg["taskType"])
	}
	if clusterCfg["dataFormat"] != "json" {
		t.Fatalf("unexpected dataFormat: %v", clusterCfg["dataFormat"])
	}
	if clusterCfg["projectId"] != "project-1" {
		t.Fatalf("unexpected projectId: %v", clusterCfg["projectId"])
	}
}

func TestIntegrationTask_CreateOrUpdateJobManagerConfig_UpdateExisting(t *testing.T) {
	existing := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "job-manager-config-tenant-1",
			Namespace: "ns-test",
		},
		Data: map[string]string{
			configFileName: `{
	  "tenants": {
	    "tenant-1": {
	      "cluster-old": {
	        "frequency": "5m",
	        "taskType": "slowquery",
	        "dataFormat": "json",
	        "projectId": "project-old"
	      }
	    }
	  }
	}`,
		},
	}

	it := setupIntegrationTaskWithFakeClient(t, existing)

	task := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "itask-2",
			Namespace: "ns-test",
			Labels: map[string]string{
				constants.LabelProjectID: "project-new",
			},
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			ClusterID:  "cluster-new",
			Frequency:  "15m",
			TaskType:   observabilitypingcapcomv1.SlowQueryTask,
			DataFormat: "parquet",
		},
	}

	err := it.createOrUpdateJobManagerConfig(context.Background(), task, "ns-test", "tenant-1")
	if err != nil {
		t.Fatalf("createOrUpdateJobManagerConfig returned error: %v", err)
	}

	updated := &corev1.ConfigMap{}
	err = it.cli.Get(context.Background(), client.ObjectKey{Name: "job-manager-config-tenant-1", Namespace: "ns-test"}, updated)
	if err != nil {
		t.Fatalf("failed to get updated ConfigMap: %v", err)
	}

	var configData map[string]map[string]map[string]map[string]interface{}
	err = json.Unmarshal([]byte(updated.Data[configFileName]), &configData)
	if err != nil {
		t.Fatalf("failed to parse updated config json: %v", err)
	}

	if configData["tenants"]["tenant-1"]["cluster-old"] == nil {
		t.Fatalf("expected existing cluster configuration to be preserved")
	}

	newClusterCfg := configData["tenants"]["tenant-1"]["cluster-new"]
	if newClusterCfg == nil {
		t.Fatalf("expected new cluster configuration to exist")
	}
	if newClusterCfg["frequency"] != "15m" {
		t.Fatalf("unexpected frequency: %v", newClusterCfg["frequency"])
	}
	if newClusterCfg["projectId"] != "project-new" {
		t.Fatalf("unexpected projectId: %v", newClusterCfg["projectId"])
	}
}

func TestIntegrationTask_CreateOrUpdateJobManagerConfig_InvalidExistingJSON(t *testing.T) {
	existing := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "job-manager-config-tenant-1",
			Namespace: "ns-test",
		},
		Data: map[string]string{
			configFileName: "{invalid-json}",
		},
	}

	it := setupIntegrationTaskWithFakeClient(t, existing)

	task := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "itask-3",
			Namespace: "ns-test",
			Labels: map[string]string{
				constants.LabelProjectID: "project-1",
			},
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			ClusterID:  "cluster-1",
			Frequency:  "10m",
			TaskType:   observabilitypingcapcomv1.SlowQueryTask,
			DataFormat: "json",
		},
	}

	err := it.createOrUpdateJobManagerConfig(context.Background(), task, "ns-test", "tenant-1")
	if err == nil {
		t.Fatalf("expected error for invalid existing config json")
	}
	if !strings.Contains(err.Error(), "failed to parse existing config") {
		t.Fatalf("unexpected error: %v", err)
	}
}

func TestIntegrationTask_CreateOrUpdateJobManagerConfig_GetError(t *testing.T) {
	it := setupIntegrationTaskWithFakeClient(t)
	it.cli = &getErrorClient{
		Client: it.cli,
		err:    errors.New("boom"),
	}

	task := &observabilitypingcapcomv1.IntegrationTask{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "itask-4",
			Namespace: "ns-test",
		},
		Spec: observabilitypingcapcomv1.IntegrationTaskSpec{
			ClusterID:  "cluster-1",
			Frequency:  "10m",
			TaskType:   observabilitypingcapcomv1.SlowQueryTask,
			DataFormat: "json",
		},
	}

	err := it.createOrUpdateJobManagerConfig(context.Background(), task, "ns-test", "tenant-1")
	if err == nil {
		t.Fatalf("expected error when Get fails")
	}
	if !strings.Contains(err.Error(), "failed to get ConfigMap") {
		t.Fatalf("unexpected error: %v", err)
	}
}