package bucket

import (
	"context"
	"fmt"
	"net"
	"testing"

	"buf.build/go/protovalidate"
	protovalidate_middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/protovalidate"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	buckettask "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task"
	infraScheme "github.com/tidbcloud/infra-provider/regional-server/pkg/scheme"
	"github.com/tidbcloud/infra-provider/regional-server/test/mockrepository"
	"go.uber.org/mock/gomock"
	"google.golang.org/genproto/googleapis/rpc/errdetails"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

func init() {
	_ = v1alpha1.AddToScheme(infraScheme.Scheme)
}

type serviceMocks struct {
	dataPlaneRW *mockrepository.MockDataPlaneReaderWriter
	appBoxRW    *mockrepository.MockAPPBoxReaderWriter
	bucketRW    *mockrepository.MockBucketReaderWriter
}

type bucketQueryerStub struct {
	status buckettask.BucketDataplanePermissionStatus
	calls  int
}

func (q *bucketQueryerStub) GetDataplanePermissionStatus(context.Context, string, string, string) (*buckettask.BucketDataplanePermissionStatus, error) {
	q.calls++
	return &q.status, nil
}

func (q *bucketQueryerStub) GetLifecycleRulesStatus(context.Context, *v1alpha1.Bucket) (*buckettask.BucketLifecycleRulesStatus, error) {
	return nil, nil
}

func (q *bucketQueryerStub) GetBucketAccessPath(context.Context, *v1alpha1.Bucket) (string, error) {
	return "", nil
}

func newServiceMocks(t *testing.T) *serviceMocks {
	t.Helper()
	ctrl := gomock.NewController(t)
	t.Cleanup(ctrl.Finish)

	return &serviceMocks{
		dataPlaneRW: mockrepository.NewMockDataPlaneReaderWriter(ctrl),
		appBoxRW:    mockrepository.NewMockAPPBoxReaderWriter(ctrl),
		bucketRW:    mockrepository.NewMockBucketReaderWriter(ctrl),
	}
}

func newBucketService(mocks *serviceMocks) *BucketService {
	return newBucketServiceWithConfig(mocks, config.RootConfig{
		Namespace: "default",
		AdditionalCloudTags: map[string]string{
			"env": "test",
		},
	})
}

func newBucketServiceWithConfig(mocks *serviceMocks, cfg config.RootConfig) *BucketService {
	return NewBucketService(
		mocks.dataPlaneRW,
		mocks.appBoxRW,
		mocks.bucketRW,
		nil,
		cfg,
	).(*BucketService)
}

func expectBucketGet(mocks *serviceMocks, ownership *v1alpha1.ResourceOwnership) {
	mocks.bucketRW.EXPECT().Get(gomock.Any(), types.NamespacedName{
		Namespace: "default",
		Name:      getBucketName("dp-123", "bucket-abc"),
	}, gomock.Any()).DoAndReturn(func(_ any, _ types.NamespacedName, bucket *v1alpha1.Bucket) error {
		bucket.Spec.Ownership = ownership
		return nil
	})
}

func newBucketReq(dataplaneID, bucketID, name, region string, tags map[string]string) *dmpbv1.CreateBucketRequest {
	if tags == nil {
		tags = make(map[string]string)
	}
	if _, ok := tags[consts.TenantLabelKey]; !ok {
		tags[consts.TenantLabelKey] = "tenant-123"
	}
	return &dmpbv1.CreateBucketRequest{
		DataplaneId: dataplaneID,
		Bucket: &dmpbv1.Bucket{
			Id:     bucketID,
			Name:   name,
			Region: region,
			Tags:   tags,
		},
	}
}

func newTestGRPCServer(t *testing.T) *grpc.Server {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	opts := []grpc.ServerOption{
		grpc.ChainUnaryInterceptor(
			protovalidate_middleware.UnaryServerInterceptor(validator),
		),
	}

	srv := grpc.NewServer(opts...)

	bucketService := &BucketService{
		config: config.RootConfig{
			Namespace: "default",
			ControllerConfig: config.ControllerConfig{
				RegionConfig: config.RegionConfig{
					Name: "us-west-2",
				},
			},
		},
	}

	dmpbv1.RegisterBucketServiceServer(srv, bucketService)

	return srv
}

func newTestGRPCClient(t *testing.T, srv *grpc.Server) dmpbv1.BucketServiceClient {
	lis := bufconn.Listen(1024 * 1024)

	go func() {
		_ = srv.Serve(lis)
	}()

	dialer := grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
		return lis.Dial()
	})

	conn, err := grpc.Dial("bufnet", dialer, grpc.WithInsecure())
	require.NoError(t, err)

	return dmpbv1.NewBucketServiceClient(conn)
}

func TestApplyLifecycleRulesValidationAllowsDisabledRule(t *testing.T) {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	err = validator.Validate(&dmpbv1.ApplyLifecycleRulesRequest{
		DataplaneId: "dp-123",
		BucketId:    "bucket-abc",
		Rules: []*dmpbv1.LifecycleRule{
			{Id: "backup-expiration", Enabled: false},
		},
	})
	require.NoError(t, err)
}

func TestBucketService_Protovalidate_InvalidRequests(t *testing.T) {
	srv := newTestGRPCServer(t)
	client := newTestGRPCClient(t, srv)

	t.Run("empty DataplaneId in CreateBucketRequest", func(t *testing.T) {
		req := &dmpbv1.CreateBucketRequest{
			DataplaneId: "",
			Bucket: &dmpbv1.Bucket{
				Id:   "bucket-abc",
				Name: "valid-name",
			},
		}

		resp, err := client.CreateBucket(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)

		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())

		fmt.Printf("Error message: %s\n", st.Message())

		for _, detail := range st.Details() {
			switch d := detail.(type) {
			case *errdetails.BadRequest:
				fmt.Println("Detail is BadRequest:")
				for _, violation := range d.FieldViolations {
					fmt.Printf(" - Field: %s, Description: %s\n", violation.Field, violation.Description)
				}
			default:
				fmt.Printf("Unknown detail type: %T\n", d)
			}
		}
	})

	t.Run("missing Bucket field in CreateBucketRequest", func(t *testing.T) {
		req := &dmpbv1.CreateBucketRequest{
			DataplaneId: "dp-123",
			Bucket:      nil,
		}

		resp, err := client.CreateBucket(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)

		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())

		fmt.Printf("Error message: %s\n", st.Message())

		for _, detail := range st.Details() {
			switch d := detail.(type) {
			case *errdetails.BadRequest:
				fmt.Println("Detail is BadRequest:")
				for _, violation := range d.FieldViolations {
					fmt.Printf(" - Field: %s, Description: %s\n", violation.Field, violation.Description)
				}
			default:
				fmt.Printf("Unknown detail type: %T\n", d)
			}
		}
	})

	t.Run("empty DataplaneId in GetBucket", func(t *testing.T) {
		req := &dmpbv1.GetBucketRequest{
			DataplaneId: "",
			BucketId:    "bucket-abc",
		}

		resp, err := client.GetBucket(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)

		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())

		fmt.Printf("Error message: %s\n", st.Message())

		for _, detail := range st.Details() {
			switch d := detail.(type) {
			case *errdetails.BadRequest:
				fmt.Println("Detail is BadRequest:")
				for _, violation := range d.FieldViolations {
					fmt.Printf(" - Field: %s, Description: %s\n", violation.Field, violation.Description)
				}
			default:
				fmt.Printf("Unknown detail type: %T\n", d)
			}
		}
	})

	t.Run("empty BucketId in GetBucket", func(t *testing.T) {
		req := &dmpbv1.GetBucketRequest{
			DataplaneId: "dp-123",
			BucketId:    "",
		}

		resp, err := client.GetBucket(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)

		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())

		fmt.Printf("Error message: %s\n", st.Message())

		for _, detail := range st.Details() {
			switch d := detail.(type) {
			case *errdetails.BadRequest:
				fmt.Println("Detail is BadRequest:")
				for _, violation := range d.FieldViolations {
					fmt.Printf(" - Field: %s, Description: %s\n", violation.Field, violation.Description)
				}
			default:
				fmt.Printf("Unknown detail type: %T\n", d)
			}
		}
	})

	t.Run("empty DataplaneId in ListBuckets", func(t *testing.T) {
		req := &dmpbv1.ListBucketsRequest{
			DataplaneId: "",
		}

		resp, err := client.ListBuckets(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)

		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())

		fmt.Printf("Error message: %s\n", st.Message())

		for _, detail := range st.Details() {
			switch d := detail.(type) {
			case *errdetails.BadRequest:
				fmt.Println("Detail is BadRequest:")
				for _, violation := range d.FieldViolations {
					fmt.Printf(" - Field: %s, Description: %s\n", violation.Field, violation.Description)
				}
			default:
				fmt.Printf("Unknown detail type: %T\n", d)
			}
		}
	})

	t.Run("empty DataplaneId in DeleteBucket", func(t *testing.T) {
		req := &dmpbv1.DeleteBucketRequest{
			DataplaneId: "",
			BucketId:    "bucket-abc",
		}

		resp, err := client.DeleteBucket(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)

		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())

		fmt.Printf("Error message: %s\n", st.Message())

		for _, detail := range st.Details() {
			switch d := detail.(type) {
			case *errdetails.BadRequest:
				fmt.Println("Detail is BadRequest:")
				for _, violation := range d.FieldViolations {
					fmt.Printf(" - Field: %s, Description: %s\n", violation.Field, violation.Description)
				}
			default:
				fmt.Printf("Unknown detail type: %T\n", d)
			}
		}
	})

	t.Run("empty BucketId in DeleteBucket", func(t *testing.T) {
		req := &dmpbv1.DeleteBucketRequest{
			DataplaneId: "dp-123",
			BucketId:    "",
		}

		resp, err := client.DeleteBucket(context.Background(), req)

		assert.Error(t, err)
		assert.Nil(t, resp)

		st, ok := status.FromError(err)
		assert.True(t, ok)
		assert.Equal(t, codes.InvalidArgument, st.Code())

		fmt.Printf("Error message: %s\n", st.Message())

		for _, detail := range st.Details() {
			switch d := detail.(type) {
			case *errdetails.BadRequest:
				fmt.Println("Detail is BadRequest:")
				for _, violation := range d.FieldViolations {
					fmt.Printf(" - Field: %s, Description: %s\n", violation.Field, violation.Description)
				}
			default:
				fmt.Printf("Unknown detail type: %T\n", d)
			}
		}
	})
}

func TestBucketService_CreateBucket_Success(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketServiceWithConfig(mocks, config.RootConfig{
		Namespace: "default",
		ResourceLabels: map[string]string{
			"tags.tidbcloud.com/servicetype": "dedicated",
		},
	})

	dataplane := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "dp-123",
			Labels: map[string]string{
				consts.TenantLabelKey: "tenant-123",
			},
		},
		Spec: v1alpha1.DataPlaneSpec{
			CloudTags: map[string]string{
				"tags.tidbcloud.com/servicetype": "byoc",
			},
			AccountRef: v1alpha1.AccountReference{
				ProviderConfigName: "1234",
			},
		},
		Status: v1alpha1.DataPlaneStatus{
			AppBox: &v1alpha1.Reference{
				Name: "appbox-123",
			},
		},
	}
	mocks.dataPlaneRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.DataPlane) {
			*out = *dataplane
		},
	).Return(nil)

	appbox := &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "appbox-123",
		},
		Spec: v1alpha1.APPBoxSpec{
			Provider: consts.ProviderTypeEKS,
		},
		Status: v1alpha1.APPBoxStatus{
			IAM: v1alpha1.IAMStatus{
				NodeRoleARN: "arn:aws:iam::1234:role/appbox-123",
			},
		},
	}
	mocks.appBoxRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.APPBox) {
			*out = *appbox
		},
	).Return(nil)

	var created *v1alpha1.Bucket
	mocks.bucketRW.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ any, bucket *v1alpha1.Bucket) error {
			created = bucket.DeepCopy()
			return nil
		},
	)

	req := newBucketReq("dp-123", "bucket-abc", "test-bucket", "us-west-2", map[string]string{
		"tags.tidbcloud.com/servicetype": "premium",
	})
	resp, err := svc.CreateBucket(context.TODO(), req)
	assert.NoError(t, err)
	assert.Equal(t, "bucket-abc", resp.Id)
	assert.Equal(t, "Pending", resp.Status)

	require.NotNil(t, created)
	assert.Equal(t, "default", created.Namespace)
	assert.Equal(t, "dp-123-bucket-abc", created.Name)
	assert.Equal(t, "dp-123", created.Spec.DataplaneId)
	assert.Equal(t, "bucket-abc", created.Spec.Id)
	assert.Equal(t, "us-west-2", created.Spec.Region)
	assert.Contains(t, created.Labels, consts.TenantLabelKey)
	assert.Equal(t, "byoc", created.Labels["tags.tidbcloud.com/servicetype"])
}

func TestBucketService_CreateBucket_MissingTenantTag(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)

	dataplane := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "dp-123",
		},
		Spec: v1alpha1.DataPlaneSpec{
			AccountRef: v1alpha1.AccountReference{
				ProviderConfigName: "1234",
			},
		},
		Status: v1alpha1.DataPlaneStatus{
			AppBox: &v1alpha1.Reference{
				Name: "appbox-123",
			},
		},
	}
	mocks.dataPlaneRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.DataPlane) {
			*out = *dataplane
		},
	).Return(nil)

	appbox := &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "appbox-123",
		},
		Spec: v1alpha1.APPBoxSpec{
			Provider: consts.ProviderTypeEKS,
		},
		Status: v1alpha1.APPBoxStatus{
			IAM: v1alpha1.IAMStatus{
				NodeRoleARN: "arn:aws:iam::1234:role/appbox-123",
			},
		},
	}
	mocks.appBoxRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.APPBox) {
			*out = *appbox
		},
	).Return(nil)

	req := newBucketReq("dp-123", "bucket-abc", "test-bucket", "us-west-2", map[string]string{
		"foo": "bar",
	})
	delete(req.Bucket.Tags, consts.TenantLabelKey)

	resp, err := svc.CreateBucket(context.TODO(), req)

	assert.Error(t, err)
	assert.Nil(t, resp)
	assert.Contains(t, err.Error(), "missing required tag: tags.tidbcloud.com/tenant")
}

func TestBucketService_CreateBucket_EmptyTags(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)

	dataplane := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "dp-123",
		},
		Spec: v1alpha1.DataPlaneSpec{
			AccountRef: v1alpha1.AccountReference{
				ProviderConfigName: "1234",
			},
		},
		Status: v1alpha1.DataPlaneStatus{
			AppBox: &v1alpha1.Reference{
				Name: "appbox-123",
			},
		},
	}
	mocks.dataPlaneRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.DataPlane) {
			*out = *dataplane
		},
	).Return(nil)

	appbox := &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "appbox-123",
		},
		Spec: v1alpha1.APPBoxSpec{
			Provider: consts.ProviderTypeEKS,
		},
		Status: v1alpha1.APPBoxStatus{
			IAM: v1alpha1.IAMStatus{
				NodeRoleARN: "arn:aws:iam::1234:role/appbox-123",
			},
		},
	}
	mocks.appBoxRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.APPBox) {
			*out = *appbox
		},
	).Return(nil)

	req := newBucketReq("dp-123", "bucket-abc", "test-bucket", "us-west-2", map[string]string{})
	delete(req.Bucket.Tags, consts.TenantLabelKey)

	resp, err := svc.CreateBucket(context.TODO(), req)

	assert.Error(t, err)
	assert.Nil(t, resp)
	assert.Contains(t, err.Error(), "missing required tag: tags.tidbcloud.com/tenant")
}

func TestBucketService_CreateBucket_Error(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)

	dataplane := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "dp-123",
			Labels: map[string]string{
				consts.TenantLabelKey: "tenant-123",
			},
		},
		Spec: v1alpha1.DataPlaneSpec{
			AccountRef: v1alpha1.AccountReference{
				ProviderConfigName: "1234",
			},
		},
		Status: v1alpha1.DataPlaneStatus{
			AppBox: &v1alpha1.Reference{
				Name: "appbox-123",
			},
		},
	}
	mocks.dataPlaneRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.DataPlane) {
			*out = *dataplane
		},
	).Return(nil)

	appbox := &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "appbox-123",
		},
		Spec: v1alpha1.APPBoxSpec{
			Provider: consts.ProviderTypeEKS,
		},
		Status: v1alpha1.APPBoxStatus{
			IAM: v1alpha1.IAMStatus{
				NodeRoleARN: "arn:aws:iam::1234:role/appbox-123",
			},
		},
	}
	mocks.appBoxRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
		func(_, _ any, out *v1alpha1.APPBox) {
			*out = *appbox
		},
	).Return(nil)

	mocks.bucketRW.EXPECT().Create(gomock.Any(), gomock.Any()).Return(fmt.Errorf("mock create error"))

	req := &dmpbv1.CreateBucketRequest{
		DataplaneId: "dp-123",
		Bucket: &dmpbv1.Bucket{
			Id:     "bucket-err",
			Name:   "test-bucket",
			Region: "us-west-2",
			Tags: map[string]string{
				consts.TenantLabelKey: "tenant-123",
			},
		},
	}

	resp, err := svc.CreateBucket(context.TODO(), req)
	assert.Error(t, err)
	assert.Nil(t, resp)
	assert.Contains(t, err.Error(), "create bucket error")
}

func TestBucketService_GetBucket_Pending(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)

	dataplaneId := "dp-123"
	bucketId := "bucket-abc"
	bucket := &v1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:      getBucketName(dataplaneId, bucketId),
			Namespace: "default",
		},
		Spec: v1alpha1.BucketSpec{
			Id:          "bucket-abc",
			DataplaneId: dataplaneId,
			Region:      "us-west-2",
		},
	}
	mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ any, key types.NamespacedName, out *v1alpha1.Bucket) error {
			*out = *bucket
			return nil
		},
	)

	resp, err := svc.GetBucket(context.TODO(), &dmpbv1.GetBucketRequest{
		DataplaneId: dataplaneId,
		BucketId:    bucketId,
	})
	assert.NoError(t, err)
	assert.Equal(t, "Pending", resp.Status)
}

func TestBucketService_GetBucket_NotFound(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)
	mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(
		apierrors.NewNotFound(schema.GroupResource{Resource: "Bucket"}, "missing"),
	)

	_, err := svc.GetBucket(context.TODO(), &dmpbv1.GetBucketRequest{
		DataplaneId: "x",
		BucketId:    "y",
	})
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "bucket not found")
}

func TestBucketService_GetBucket_K8sError(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)
	mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("k8s get fail"))

	_, err := svc.GetBucket(context.TODO(), &dmpbv1.GetBucketRequest{
		DataplaneId: "dp",
		BucketId:    "id",
	})
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "get bucket error")
}

func TestBucketService_ListBuckets_Success(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)

	bucket := &v1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "dp-123-bucket-abc",
			Namespace: "default",
			Labels: map[string]string{
				consts.DataplaneLabelKey: "dp-123",
			},
		},
		Spec: v1alpha1.BucketSpec{
			Id:     "bucket-abc",
			Region: "us-west-2",
		},
	}
	mocks.bucketRW.EXPECT().List(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ any, list *v1alpha1.BucketList, _ ...client.ListOption) error {
			list.Items = []v1alpha1.Bucket{*bucket}
			return nil
		},
	)

	resp, err := svc.ListBuckets(context.TODO(), &dmpbv1.ListBucketsRequest{
		DataplaneId: "dp-123",
	})
	assert.NoError(t, err)
	assert.Len(t, resp.Buckets, 1)
	assert.Equal(t, "Creating", resp.Buckets[0].Status)
}

func TestBucketService_ListBuckets_Error(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)
	mocks.bucketRW.EXPECT().List(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("mock list error"))

	_, err := svc.ListBuckets(context.TODO(), &dmpbv1.ListBucketsRequest{DataplaneId: "dp"})
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "list buckets error")
}

func TestBucketService_DeleteBucket_Success(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)

	bucket := &v1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "dp-123-bucket-abc",
			Namespace: "default",
		},
	}
	mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ any, key types.NamespacedName, out *v1alpha1.Bucket) error {
			*out = *bucket
			return nil
		},
	)
	mocks.bucketRW.EXPECT().Delete(gomock.Any(), gomock.Any()).Return(nil)

	_, err := svc.DeleteBucket(context.TODO(), &dmpbv1.DeleteBucketRequest{
		DataplaneId: "dp-123",
		BucketId:    "bucket-abc",
	})
	assert.NoError(t, err)
}

func TestBucketService_DeleteBucket_GetError(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)
	mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("mock get error"))

	_, err := svc.DeleteBucket(context.TODO(), &dmpbv1.DeleteBucketRequest{
		DataplaneId: "dp",
		BucketId:    "abc",
	})
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "get bucket error")
}

func TestBucketService_DeleteBucket_DeleteError(t *testing.T) {
	mocks := newServiceMocks(t)
	svc := newBucketService(mocks)
	mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ any, key types.NamespacedName, bucket *v1alpha1.Bucket) error {
			*bucket = v1alpha1.Bucket{
				ObjectMeta: metav1.ObjectMeta{
					Name:      key.Name,
					Namespace: key.Namespace,
				},
			}
			return nil
		},
	)
	mocks.bucketRW.EXPECT().Delete(gomock.Any(), gomock.Any()).Return(fmt.Errorf("mock delete error"))

	_, err := svc.DeleteBucket(context.TODO(), &dmpbv1.DeleteBucketRequest{
		DataplaneId: "dp-123",
		BucketId:    "bucket-abc",
	})
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delete bucket error")
}

func TestBucketService_GrantDataplanePermission(t *testing.T) {
	t.Run("GrantDataplanePermission Silent Success For AKS BYOC Static Auth", func(t *testing.T) {
		mocks := newServiceMocks(t)
		expectBucketGet(mocks, nil)
		svc := newBucketServiceWithConfig(mocks, config.RootConfig{
			Namespace: "default",
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeAKS,
			},
			BYOC: config.BYOCConfig{
				AKSControlPlaneIdentityName:         "control-plane",
				AKSKubeletIdentityName:              "kubelet",
				AKSManagedIdentityResourceGroupName: "identities-rg",
			},
		})

		resp, err := svc.GrantDataplanePermission(context.TODO(), &dmpbv1.GrantDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
	})

	t.Run("GrantDataplanePermission Success", func(t *testing.T) {
		mocks := newServiceMocks(t)
		svc := newBucketService(mocks)

		mocks.dataPlaneRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
			func(_, _ any, out *v1alpha1.DataPlane) {
				out.Namespace = "default"
				out.Name = "dp-456"
			},
		).Return(nil)

		bucket := &v1alpha1.Bucket{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: "default",
				Name:      getBucketName("dp-123", "bucket-abc"),
			},
			Spec: v1alpha1.BucketSpec{
				DataplaneId: "dp-123",
				Id:          "bucket-abc",
			},
		}
		mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ any, _ types.NamespacedName, out *v1alpha1.Bucket) error {
				*out = *bucket
				return nil
			},
		)
		mocks.bucketRW.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ any, in *v1alpha1.Bucket) error {
				bucket = in.DeepCopy()
				return nil
			},
		)

		_, err := svc.GrantDataplanePermission(context.TODO(), &dmpbv1.GrantDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.NoError(t, err)

		assert.Contains(t, bucket.Spec.GrantedDataplaneIds, "dp-456")
	})

	t.Run("GrantDataplanePermission Already Exists", func(t *testing.T) {
		mocks := newServiceMocks(t)
		svc := newBucketService(mocks)

		mocks.dataPlaneRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Do(
			func(_, _ any, out *v1alpha1.DataPlane) {
				out.Namespace = "default"
				out.Name = "dp-456"
			},
		).Return(nil).Times(2)

		bucket := &v1alpha1.Bucket{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: "default",
				Name:      getBucketName("dp-123", "bucket-abc"),
			},
			Spec: v1alpha1.BucketSpec{
				DataplaneId:         "dp-123",
				Id:                  "bucket-abc",
				GrantedDataplaneIds: []string{"dp-456"},
			},
		}
		mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ any, _ types.NamespacedName, out *v1alpha1.Bucket) error {
				*out = *bucket
				return nil
			},
		).Times(2)

		_, err := svc.GrantDataplanePermission(context.TODO(), &dmpbv1.GrantDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.NoError(t, err)

		_, err = svc.GrantDataplanePermission(context.TODO(), &dmpbv1.GrantDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.NoError(t, err) // Should not error, just return early

		assert.Contains(t, bucket.Spec.GrantedDataplaneIds, "dp-456")
	})

	t.Run("Get Bucket Error", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		mockdataplaneRW := mockrepository.NewMockDataPlaneReaderWriter(ctrl)
		mockbucketRW := mockrepository.NewMockBucketReaderWriter(ctrl)
		mockbucketRW.EXPECT().Get(gomock.Any(), types.NamespacedName{
			Namespace: "default",
			Name:      getBucketName("dp-123", "bucket-abc"),
		}, gomock.Any()).Return(fmt.Errorf("mock k8s error"))

		svc := &BucketService{
			dataPlaneRW: mockdataplaneRW,
			bucketRW:    mockbucketRW,
			config: config.RootConfig{
				Namespace: "default",
			},
		}

		_, err := svc.GrantDataplanePermission(context.TODO(), &dmpbv1.GrantDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.Error(t, err)
		assert.Contains(t, err.Error(), "get bucket error")
	})

	t.Run("K8s Client Error Getting Permitted Dataplane", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()

		// Mock dataplane RW to return a generic error
		mockdataplaneRW := mockrepository.NewMockDataPlaneReaderWriter(ctrl)
		mockdataplaneRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("mock k8s error"))

		// Mock appbox RW (not used in this test but required by service)
		mockappboxRW := mockrepository.NewMockAPPBoxReaderWriter(ctrl)
		mockbucketRW := mockrepository.NewMockBucketReaderWriter(ctrl)
		mockbucketRW.EXPECT().Get(gomock.Any(), types.NamespacedName{
			Namespace: "default",
			Name:      getBucketName("dp-123", "bucket-abc"),
		}, gomock.Any()).Return(nil)

		svc := &BucketService{
			dataPlaneRW: mockdataplaneRW,
			appBoxRW:    mockappboxRW,
			bucketRW:    mockbucketRW,
			config: config.RootConfig{
				Namespace: "default",
			},
		}

		_, err := svc.GrantDataplanePermission(context.TODO(), &dmpbv1.GrantDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.Error(t, err)
		assert.Contains(t, err.Error(), "get permitted dataplane error")
	})
}

func TestBucketService_RevokeDataplanePermission(t *testing.T) {
	t.Run("RevokeDataplanePermission Silent Success For AKS BYOC Static Auth", func(t *testing.T) {
		mocks := newServiceMocks(t)
		expectBucketGet(mocks, nil)
		svc := newBucketServiceWithConfig(mocks, config.RootConfig{
			Namespace: "default",
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeAKS,
			},
			BYOC: config.BYOCConfig{
				AKSControlPlaneIdentityName:         "control-plane",
				AKSKubeletIdentityName:              "kubelet",
				AKSManagedIdentityResourceGroupName: "identities-rg",
			},
		})

		resp, err := svc.RevokeDataplanePermission(context.TODO(), &dmpbv1.RevokeDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
	})

	t.Run("RevokeDataplanePermission Success", func(t *testing.T) {
		mocks := newServiceMocks(t)
		svc := newBucketService(mocks)

		// Create a bucket with existing granted dataplane IDs
		bucket := &v1alpha1.Bucket{
			ObjectMeta: metav1.ObjectMeta{
				Name:      getBucketName("dp-123", "bucket-abc"),
				Namespace: "default",
			},
			Spec: v1alpha1.BucketSpec{
				DataplaneId:         "dp-123",
				Id:                  "bucket-abc",
				GrantedDataplaneIds: []string{"dp-456", "dp-789"},
			},
		}
		mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ any, _ types.NamespacedName, out *v1alpha1.Bucket) error {
				*out = *bucket
				return nil
			},
		)
		mocks.bucketRW.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ any, in *v1alpha1.Bucket) error {
				bucket = in.DeepCopy()
				return nil
			},
		)

		// Revoke the permission
		_, err := svc.RevokeDataplanePermission(context.TODO(), &dmpbv1.RevokeDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.NoError(t, err)

		// Verify the permission was removed
		require.NotNil(t, bucket)
		assert.NotContains(t, bucket.Spec.GrantedDataplaneIds, "dp-456")
		assert.Contains(t, bucket.Spec.GrantedDataplaneIds, "dp-789")
	})

	t.Run("RevokeDataplanePermission Permission Not Found", func(t *testing.T) {
		mocks := newServiceMocks(t)
		svc := newBucketService(mocks)

		// Create a bucket with existing granted dataplane IDs
		bucket := &v1alpha1.Bucket{
			ObjectMeta: metav1.ObjectMeta{
				Name:      getBucketName("dp-123", "bucket-abc"),
				Namespace: "default",
			},
			Spec: v1alpha1.BucketSpec{
				DataplaneId:         "dp-123",
				Id:                  "bucket-abc",
				GrantedDataplaneIds: []string{"dp-456"},
			},
		}
		mocks.bucketRW.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ any, _ types.NamespacedName, out *v1alpha1.Bucket) error {
				*out = *bucket
				return nil
			},
		)

		// Try to revoke non-existent permission
		_, err := svc.RevokeDataplanePermission(context.TODO(), &dmpbv1.RevokeDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-nonexistent",
		})
		require.NoError(t, err) // Should not error, just return early

		// Verify the original permission is still there
		require.NotNil(t, bucket)
		assert.Contains(t, bucket.Spec.GrantedDataplaneIds, "dp-456")
	})

	t.Run("Get Bucket Error", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()

		mockbucketRW := mockrepository.NewMockBucketReaderWriter(ctrl)
		mockbucketRW.EXPECT().Get(gomock.Any(), types.NamespacedName{
			Namespace: "default",
			Name:      getBucketName("dp-123", "bucket-abc"),
		}, gomock.Any()).Return(fmt.Errorf("mock k8s error"))

		svc := &BucketService{
			bucketRW: mockbucketRW,
			config: config.RootConfig{
				Namespace: "default",
			},
		}

		_, err := svc.RevokeDataplanePermission(context.TODO(), &dmpbv1.RevokeDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.Error(t, err)
		assert.Contains(t, err.Error(), "get bucket error")
	})
}

func TestBucketService_GetDataplanePermission(t *testing.T) {
	t.Run("GetDataplanePermission Returns Synced For AKS BYOC Static Auth", func(t *testing.T) {
		mocks := newServiceMocks(t)
		expectBucketGet(mocks, nil)
		svc := newBucketServiceWithConfig(mocks, config.RootConfig{
			Namespace: "default",
			ControllerConfig: config.ControllerConfig{
				Provider: consts.ProviderTypeAKS,
			},
			BYOC: config.BYOCConfig{
				AKSControlPlaneIdentityName:         "control-plane",
				AKSKubeletIdentityName:              "kubelet",
				AKSManagedIdentityResourceGroupName: "identities-rg",
			},
		})

		resp, err := svc.GetDataplanePermission(context.TODO(), &dmpbv1.GetDataplanePermissionRequest{
			DataplaneId:          "dp-123",
			BucketId:             "bucket-abc",
			PermittedDataplaneId: "dp-456",
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
		assert.Equal(t, dmpbv1.GetDataplanePermissionResponse_Synced, resp.Status)
	})
}

func TestBucketService_GetDataplanePermissionOwnership(t *testing.T) {
	tests := []struct {
		name             string
		currentOwnership *v1alpha1.ResourceOwnership
		requestOwnership *dmpbv1.ResourceOwnershipContext
		wantConflict     bool
		wantQuery        bool
	}{
		{
			name:             "ownerless bucket remains compatible",
			requestOwnership: &dmpbv1.ResourceOwnershipContext{SharedPoolId: "sp-2"},
			wantQuery:        true,
		},
		{
			name:             "matching owner",
			currentOwnership: &v1alpha1.ResourceOwnership{SharedPoolID: "sp-1"},
			requestOwnership: &dmpbv1.ResourceOwnershipContext{SharedPoolId: "sp-1"},
			wantQuery:        true,
		},
		{
			name:             "missing request owner",
			currentOwnership: &v1alpha1.ResourceOwnership{SharedPoolID: "sp-1"},
			wantConflict:     true,
		},
		{
			name:             "different request owner",
			currentOwnership: &v1alpha1.ResourceOwnership{SharedPoolID: "sp-1"},
			requestOwnership: &dmpbv1.ResourceOwnershipContext{SharedPoolId: "sp-2"},
			wantConflict:     true,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			mocks := newServiceMocks(t)
			expectBucketGet(mocks, test.currentOwnership)
			queryer := &bucketQueryerStub{status: buckettask.BucketDataplanePermissionStatusSynced}
			svc := newBucketService(mocks)
			svc.bucketQueryer = queryer

			resp, err := svc.GetDataplanePermission(context.TODO(), &dmpbv1.GetDataplanePermissionRequest{
				DataplaneId:          "dp-123",
				BucketId:             "bucket-abc",
				PermittedDataplaneId: "dp-456",
				OwnershipContext:     test.requestOwnership,
			})
			if test.wantConflict {
				require.True(t, bizerr.Is(err, bizerr.ResourceConflict))
				require.Nil(t, resp)
			} else {
				require.NoError(t, err)
				require.Equal(t, dmpbv1.GetDataplanePermissionResponse_Synced, resp.Status)
			}
			if test.wantQuery {
				require.Equal(t, 1, queryer.calls)
			} else {
				require.Zero(t, queryer.calls)
			}
		})
	}
}

func TestBucketService_AKSStaticAuthRejectsMismatchedPermissionOwnership(t *testing.T) {
	tests := []struct {
		name string
		call func(*BucketService) error
	}{
		{
			name: "grant",
			call: func(svc *BucketService) error {
				_, err := svc.GrantDataplanePermission(context.TODO(), &dmpbv1.GrantDataplanePermissionRequest{
					DataplaneId:          "dp-123",
					BucketId:             "bucket-abc",
					PermittedDataplaneId: "dp-456",
					OwnershipContext:     &dmpbv1.ResourceOwnershipContext{SharedPoolId: "sp-2"},
				})
				return err
			},
		},
		{
			name: "revoke",
			call: func(svc *BucketService) error {
				_, err := svc.RevokeDataplanePermission(context.TODO(), &dmpbv1.RevokeDataplanePermissionRequest{
					DataplaneId:          "dp-123",
					BucketId:             "bucket-abc",
					PermittedDataplaneId: "dp-456",
					OwnershipContext:     &dmpbv1.ResourceOwnershipContext{SharedPoolId: "sp-2"},
				})
				return err
			},
		},
		{
			name: "get",
			call: func(svc *BucketService) error {
				_, err := svc.GetDataplanePermission(context.TODO(), &dmpbv1.GetDataplanePermissionRequest{
					DataplaneId:          "dp-123",
					BucketId:             "bucket-abc",
					PermittedDataplaneId: "dp-456",
					OwnershipContext:     &dmpbv1.ResourceOwnershipContext{SharedPoolId: "sp-2"},
				})
				return err
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			mocks := newServiceMocks(t)
			expectBucketGet(mocks, &v1alpha1.ResourceOwnership{SharedPoolID: "sp-1"})
			svc := newBucketServiceWithConfig(mocks, config.RootConfig{
				Namespace: "default",
				ControllerConfig: config.ControllerConfig{
					Provider: consts.ProviderTypeAKS,
				},
				BYOC: config.BYOCConfig{
					AKSControlPlaneIdentityName:         "control-plane",
					AKSKubeletIdentityName:              "kubelet",
					AKSManagedIdentityResourceGroupName: "identities-rg",
				},
			})

			require.True(t, bizerr.Is(test.call(svc), bizerr.ResourceConflict))
		})
	}
}

func TestConvertLifecycleRules2CR(t *testing.T) {
	tests := []struct {
		name     string
		input    []*dmpbv1.LifecycleRule
		expected []v1alpha1.BucketLifecycleRule
	}{
		{
			name:     "empty input",
			input:    []*dmpbv1.LifecycleRule{},
			expected: []v1alpha1.BucketLifecycleRule{},
		},
		{
			name: "single rule with all fields",
			input: []*dmpbv1.LifecycleRule{
				{
					Id:                             "test-rule-1",
					Enabled:                        true,
					PrefixFilter:                   "logs/",
					ExpirationDays:                 30,
					IncompleteUploadExpirationDays: 7,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "env", Value: "production"},
						{Key: "team", Value: "backend"},
					},
				},
			},
			expected: []v1alpha1.BucketLifecycleRule{
				{
					Id:                             "test-rule-1",
					Enabled:                        true,
					PrefixFilter:                   "logs/",
					ExpirationDays:                 30,
					IncompleteUploadExpirationDays: 7,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "env", Value: "production"},
						{Key: "team", Value: "backend"},
					},
				},
			},
		},
		{
			name: "multiple rules with different configurations",
			input: []*dmpbv1.LifecycleRule{
				{
					Id:             "rule-1",
					Enabled:        true,
					PrefixFilter:   "documents/",
					ExpirationDays: 90,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "type", Value: "document"},
					},
				},
				{
					Id:                             "rule-2",
					Enabled:                        false,
					IncompleteUploadExpirationDays: 14,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "category", Value: "temporary"},
						{Key: "importance", Value: "low"},
					},
				},
				{
					Id:             "rule-3",
					Enabled:        true,
					ExpirationDays: 365,
					// No prefix filter or tags filter
				},
			},
			expected: []v1alpha1.BucketLifecycleRule{
				{
					Id:             "rule-1",
					Enabled:        true,
					PrefixFilter:   "documents/",
					ExpirationDays: 90,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "type", Value: "document"},
					},
				},
				{
					Id:                             "rule-2",
					Enabled:                        false,
					IncompleteUploadExpirationDays: 14,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "category", Value: "temporary"},
						{Key: "importance", Value: "low"},
					},
				},
				{
					Id:             "rule-3",
					Enabled:        true,
					ExpirationDays: 365,
					TagsFilter:     []v1alpha1.BucketTagFilter{}, // empty slice
				},
			},
		},
		{
			name: "rule with no tags filter",
			input: []*dmpbv1.LifecycleRule{
				{
					Id:             "simple-rule",
					Enabled:        true,
					PrefixFilter:   "archive/",
					ExpirationDays: 180,
					// No TagsFilter
				},
			},
			expected: []v1alpha1.BucketLifecycleRule{
				{
					Id:             "simple-rule",
					Enabled:        true,
					PrefixFilter:   "archive/",
					ExpirationDays: 180,
					TagsFilter:     []v1alpha1.BucketTagFilter{}, // empty slice
				},
			},
		},
		{
			name: "rule with only tags filter, no prefix",
			input: []*dmpbv1.LifecycleRule{
				{
					Id:             "tags-only-rule",
					Enabled:        true,
					ExpirationDays: 60,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "env", Value: "staging"},
						{Key: "auto-delete", Value: "true"},
					},
					// No PrefixFilter
				},
			},
			expected: []v1alpha1.BucketLifecycleRule{
				{
					Id:             "tags-only-rule",
					Enabled:        true,
					ExpirationDays: 60,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "env", Value: "staging"},
						{Key: "auto-delete", Value: "true"},
					},
					// PrefixFilter should be empty string (default)
				},
			},
		},
		{
			name: "rule with zero expiration days",
			input: []*dmpbv1.LifecycleRule{
				{
					Id:                             "zero-expiration-rule",
					Enabled:                        true,
					PrefixFilter:                   "uploads/incomplete/",
					ExpirationDays:                 0, // zero value
					IncompleteUploadExpirationDays: 1,
				},
			},
			expected: []v1alpha1.BucketLifecycleRule{
				{
					Id:                             "zero-expiration-rule",
					Enabled:                        true,
					PrefixFilter:                   "uploads/incomplete/",
					ExpirationDays:                 0, // zero value preserved
					IncompleteUploadExpirationDays: 1,
					TagsFilter:                     []v1alpha1.BucketTagFilter{}, // empty slice
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := convertLifecycleRules2CR(tt.input)
			assert.Equal(t, tt.expected, result)

			// Validate length
			assert.Equal(t, len(tt.input), len(result))

			// Validate individual fields for each rule
			for i, expectedRule := range tt.expected {
				if i < len(result) {
					actualRule := result[i]
					assert.Equal(t, expectedRule.Id, actualRule.Id, "Rule ID mismatch at index %d", i)
					assert.Equal(t, expectedRule.Enabled, actualRule.Enabled, "Rule Enabled mismatch at index %d", i)
					assert.Equal(t, expectedRule.PrefixFilter, actualRule.PrefixFilter, "Rule PrefixFilter mismatch at index %d", i)
					assert.Equal(t, expectedRule.ExpirationDays, actualRule.ExpirationDays, "Rule ExpirationDays mismatch at index %d", i)
					assert.Equal(t, expectedRule.IncompleteUploadExpirationDays, actualRule.IncompleteUploadExpirationDays, "Rule IncompleteUploadExpirationDays mismatch at index %d", i)
					assert.Equal(t, len(expectedRule.TagsFilter), len(actualRule.TagsFilter), "Rule TagsFilter length mismatch at index %d", i)

					// Validate tags filter
					for j, expectedTag := range expectedRule.TagsFilter {
						if j < len(actualRule.TagsFilter) {
							actualTag := actualRule.TagsFilter[j]
							assert.Equal(t, expectedTag.Key, actualTag.Key, "Tag Key mismatch at rule %d, tag %d", i, j)
							assert.Equal(t, expectedTag.Value, actualTag.Value, "Tag Value mismatch at rule %d, tag %d", i, j)
						}
					}
				}
			}
		})
	}
}

func TestConvertLifecycleRules2CR_NilInput(t *testing.T) {
	result := convertLifecycleRules2CR(nil)
	assert.NotNil(t, result)
	assert.Empty(t, result)
}

func TestConvertLifecycleRules2CR_PreservesOrder(t *testing.T) {
	input := []*dmpbv1.LifecycleRule{
		{Id: "first", Enabled: true, ExpirationDays: 10},
		{Id: "second", Enabled: false, ExpirationDays: 20},
		{Id: "third", Enabled: true, ExpirationDays: 30},
	}

	result := convertLifecycleRules2CR(input)

	require.Len(t, result, 3)
	assert.Equal(t, "first", result[0].Id)
	assert.Equal(t, "second", result[1].Id)
	assert.Equal(t, "third", result[2].Id)

	assert.Equal(t, uint32(10), result[0].ExpirationDays)
	assert.Equal(t, uint32(20), result[1].ExpirationDays)
	assert.Equal(t, uint32(30), result[2].ExpirationDays)
}

func TestConvertLifecycleRules2Resp(t *testing.T) {
	tests := []struct {
		name     string
		input    []v1alpha1.BucketLifecycleRule
		expected []*dmpbv1.LifecycleRule
	}{
		{
			name:     "empty input",
			input:    []v1alpha1.BucketLifecycleRule{},
			expected: []*dmpbv1.LifecycleRule{},
		},
		{
			name: "single rule with all fields",
			input: []v1alpha1.BucketLifecycleRule{
				{
					Id:                             "test-rule-1",
					Enabled:                        true,
					PrefixFilter:                   "logs/",
					ExpirationDays:                 30,
					IncompleteUploadExpirationDays: 7,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "env", Value: "production"},
						{Key: "team", Value: "backend"},
					},
				},
			},
			expected: []*dmpbv1.LifecycleRule{
				{
					Id:                             "test-rule-1",
					Enabled:                        true,
					PrefixFilter:                   "logs/",
					ExpirationDays:                 30,
					IncompleteUploadExpirationDays: 7,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "env", Value: "production"},
						{Key: "team", Value: "backend"},
					},
				},
			},
		},
		{
			name: "multiple rules with different configurations",
			input: []v1alpha1.BucketLifecycleRule{
				{
					Id:             "rule-1",
					Enabled:        true,
					PrefixFilter:   "documents/",
					ExpirationDays: 90,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "type", Value: "document"},
					},
				},
				{
					Id:                             "rule-2",
					Enabled:                        false,
					IncompleteUploadExpirationDays: 14,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "category", Value: "temporary"},
						{Key: "importance", Value: "low"},
					},
				},
				{
					Id:             "rule-3",
					Enabled:        true,
					ExpirationDays: 365,
					TagsFilter:     []v1alpha1.BucketTagFilter{}, // empty slice
				},
			},
			expected: []*dmpbv1.LifecycleRule{
				{
					Id:             "rule-1",
					Enabled:        true,
					PrefixFilter:   "documents/",
					ExpirationDays: 90,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "type", Value: "document"},
					},
				},
				{
					Id:                             "rule-2",
					Enabled:                        false,
					IncompleteUploadExpirationDays: 14,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "category", Value: "temporary"},
						{Key: "importance", Value: "low"},
					},
				},
				{
					Id:             "rule-3",
					Enabled:        true,
					ExpirationDays: 365,
					TagsFilter:     []*dmpbv1.LifecycleRule_Tag{}, // empty slice
				},
			},
		},
		{
			name: "rule with no tags filter",
			input: []v1alpha1.BucketLifecycleRule{
				{
					Id:             "simple-rule",
					Enabled:        true,
					PrefixFilter:   "archive/",
					ExpirationDays: 180,
					TagsFilter:     []v1alpha1.BucketTagFilter{}, // empty slice
				},
			},
			expected: []*dmpbv1.LifecycleRule{
				{
					Id:             "simple-rule",
					Enabled:        true,
					PrefixFilter:   "archive/",
					ExpirationDays: 180,
					TagsFilter:     []*dmpbv1.LifecycleRule_Tag{}, // empty slice
				},
			},
		},
		{
			name: "rule with only tags filter, no prefix",
			input: []v1alpha1.BucketLifecycleRule{
				{
					Id:             "tags-only-rule",
					Enabled:        true,
					ExpirationDays: 60,
					TagsFilter: []v1alpha1.BucketTagFilter{
						{Key: "env", Value: "staging"},
						{Key: "auto-delete", Value: "true"},
					},
					// PrefixFilter should be empty string (default)
				},
			},
			expected: []*dmpbv1.LifecycleRule{
				{
					Id:             "tags-only-rule",
					Enabled:        true,
					ExpirationDays: 60,
					TagsFilter: []*dmpbv1.LifecycleRule_Tag{
						{Key: "env", Value: "staging"},
						{Key: "auto-delete", Value: "true"},
					},
					// PrefixFilter should be empty string (default)
				},
			},
		},
		{
			name: "rule with zero expiration days",
			input: []v1alpha1.BucketLifecycleRule{
				{
					Id:                             "zero-expiration-rule",
					Enabled:                        true,
					PrefixFilter:                   "uploads/incomplete/",
					ExpirationDays:                 0, // zero value
					IncompleteUploadExpirationDays: 1,
					TagsFilter:                     []v1alpha1.BucketTagFilter{}, // empty slice
				},
			},
			expected: []*dmpbv1.LifecycleRule{
				{
					Id:                             "zero-expiration-rule",
					Enabled:                        true,
					PrefixFilter:                   "uploads/incomplete/",
					ExpirationDays:                 0, // zero value preserved
					IncompleteUploadExpirationDays: 1,
					TagsFilter:                     []*dmpbv1.LifecycleRule_Tag{}, // empty slice
				},
			},
		},
		{
			name: "rule with only incomplete upload expiration",
			input: []v1alpha1.BucketLifecycleRule{
				{
					Id:                             "incomplete-only-rule",
					Enabled:                        true,
					IncompleteUploadExpirationDays: 5,
					// No other fields set
				},
			},
			expected: []*dmpbv1.LifecycleRule{
				{
					Id:                             "incomplete-only-rule",
					Enabled:                        true,
					IncompleteUploadExpirationDays: 5,
					TagsFilter:                     []*dmpbv1.LifecycleRule_Tag{},
					// Other fields should be zero values
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := convertLifecycleRules2Resp(tt.input)
			assert.Equal(t, tt.expected, result)

			// Validate length
			assert.Equal(t, len(tt.input), len(result))

			// Validate individual fields for each rule
			for i, expectedRule := range tt.expected {
				if i < len(result) {
					actualRule := result[i]
					assert.Equal(t, expectedRule.Id, actualRule.Id, "Rule ID mismatch at index %d", i)
					assert.Equal(t, expectedRule.Enabled, actualRule.Enabled, "Rule Enabled mismatch at index %d", i)
					assert.Equal(t, expectedRule.PrefixFilter, actualRule.PrefixFilter, "Rule PrefixFilter mismatch at index %d", i)
					assert.Equal(t, expectedRule.ExpirationDays, actualRule.ExpirationDays, "Rule ExpirationDays mismatch at index %d", i)
					assert.Equal(t, expectedRule.IncompleteUploadExpirationDays, actualRule.IncompleteUploadExpirationDays, "Rule IncompleteUploadExpirationDays mismatch at index %d", i)
					assert.Equal(t, len(expectedRule.TagsFilter), len(actualRule.TagsFilter), "Rule TagsFilter length mismatch at index %d", i)

					// Validate tags filter
					for j, expectedTag := range expectedRule.TagsFilter {
						if j < len(actualRule.TagsFilter) {
							actualTag := actualRule.TagsFilter[j]
							assert.Equal(t, expectedTag.Key, actualTag.Key, "Tag Key mismatch at rule %d, tag %d", i, j)
							assert.Equal(t, expectedTag.Value, actualTag.Value, "Tag Value mismatch at rule %d, tag %d", i, j)
						}
					}
				}
			}
		})
	}
}