From b7725553d08bd4bed012916604c8dc14813981cd Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Wed, 1 Jul 2026 14:42:39 -0700 Subject: [PATCH 1/4] test: add a generic S3 agent builder to the object test client Add client.NewS3Agent, which builds an rgw S3 agent for an object store from a set of credentials and the store's endpoint. The object tests build S3 clients from several credential sources (the store's dashboard admin user, an OBC's provisioned secret); this factors out the endpoint and agent construction they share. Signed-off-by: Joshua Hoblitt --- tests/integration/object/util/client/s3.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/integration/object/util/client/s3.go b/tests/integration/object/util/client/s3.go index f521c9eed..ae3cc3b10 100644 --- a/tests/integration/object/util/client/s3.go +++ b/tests/integration/object/util/client/s3.go @@ -25,6 +25,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + rgw "github.com/rook/rook/pkg/operator/ceph/object" "github.com/rook/rook/tests/framework/installer" "github.com/rook/rook/tests/framework/utils" ) @@ -73,3 +74,18 @@ func GetS3Endpoint(objectStore *cephv1.CephObjectStore, k8sh *utils.K8sHelper, t return endpoint, nil } + +// NewS3Agent builds an S3 client for objectStore using the given credentials. +func NewS3Agent(objectStore *cephv1.CephObjectStore, k8sh *utils.K8sHelper, tlsEnable bool, accessKey, secretKey string) (*rgw.S3Agent, error) { + endpoint, err := GetS3Endpoint(objectStore, k8sh, tlsEnable) + if err != nil { + return nil, errors.Wrap(err, "failed to get s3 endpoint") + } + + agent, err := rgw.NewS3Agent(accessKey, secretKey, endpoint, true, nil, tlsEnable, nil) + if err != nil { + return nil, errors.Wrap(err, "failed to create s3 agent") + } + + return agent, nil +} From 2f82880de758ffe17978db36d699a59055c27315 Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Wed, 1 Jul 2026 14:42:39 -0700 Subject: [PATCH 2/4] test: consolidate the OBC test helpers into util/obc Introduce util/obc as the home for ObjectBucketClaim test helpers: the provisioner StorageClass constructor (moved out of util/fixture, where it was a misplaced pure constructor), the create/bound and delete/absent lifecycle waiters, and a per-OBC S3 client. The lifecycle waiters and S3 client are promoted from the notification suite, which built them inline as their only consumer. Retarget the StorageClass callers (topic/kafka, bucket/owner) and rework the notification suite onto the shared helpers. Signed-off-by: Joshua Hoblitt --- .../integration/object/bucket/owner/owner.go | 7 +- .../object/notification/notification.go | 116 +++++------------- tests/integration/object/topic/kafka/kafka.go | 3 +- tests/integration/object/util/fixture/obc.go | 39 ------ tests/integration/object/util/obc/obc.go | 104 ++++++++++++++++ 5 files changed, 142 insertions(+), 127 deletions(-) delete mode 100644 tests/integration/object/util/fixture/obc.go create mode 100644 tests/integration/object/util/obc/obc.go diff --git a/tests/integration/object/bucket/owner/owner.go b/tests/integration/object/bucket/owner/owner.go index 70b8cab18..b430468c6 100644 --- a/tests/integration/object/bucket/owner/owner.go +++ b/tests/integration/object/bucket/owner/owner.go @@ -34,6 +34,7 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/tests/framework/utils" "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/obc" "github.com/rook/rook/tests/integration/object/util/sharedstore" "github.com/rook/rook/tests/integration/object/util/wait4" ) @@ -52,7 +53,7 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, store }, } - storageClass = fixture.StorageClass(defaultName, objectStore) + storageClass = obc.StorageClass(defaultName, objectStore) // test user without any quotas set osu1 = cephv1.CephObjectStoreUser{ @@ -264,8 +265,8 @@ func TestObjectBucketClaimBucketOwner(t *testing.T, k8sh *utils.K8sHelper, store t.Run(fmt.Sprintf("obc %q has no bucketOwner", obc1.Name), func(t *testing.T) { // verify that bucketOwner is unset on the live obc wait4.AssertCondition(ctx, t, obcClient, obc1.Name, - func(obc *bktv1alpha1.ObjectBucketClaim) bool { - _, ok := obc.Spec.AdditionalConfig["bucketOwner"] + func(liveObc *bktv1alpha1.ObjectBucketClaim) bool { + _, ok := liveObc.Spec.AdditionalConfig["bucketOwner"] return !ok }, wait4.TimeoutShort) diff --git a/tests/integration/object/notification/notification.go b/tests/integration/object/notification/notification.go index e6a8785e3..bf99d7e99 100644 --- a/tests/integration/object/notification/notification.go +++ b/tests/integration/object/notification/notification.go @@ -37,8 +37,8 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" rgw "github.com/rook/rook/pkg/operator/ceph/object" "github.com/rook/rook/tests/framework/utils" - "github.com/rook/rook/tests/integration/object/util/client" "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/obc" "github.com/rook/rook/tests/integration/object/util/sharedstore" "github.com/rook/rook/tests/integration/object/util/wait4" ) @@ -110,11 +110,11 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst sinkSel = labels.SelectorFromSet(labels.Set{"app": sinkName}) sinkURI = fmt.Sprintf("http://%s.%s:%d", sinkName, ns.Name, sinkPort) - storageClass = fixture.StorageClass(defaultName, objectStore) + storageClass = obc.StorageClass(defaultName, objectStore) topic = httpTopic(defaultName+"-topic", ns.Name, objectStore, sinkURI) notification = bucketNotification(defaultName+"-notification", ns.Name, topic.Name) - obc = notificationOBC(defaultName+"-obc", ns.Name, storageClass.Name, notification.Name) + obc1 = notificationOBC(defaultName+"-obc", ns.Name, storageClass.Name, notification.Name) body = "test bucket notification payload" contentType = "text/plain" @@ -145,14 +145,15 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst require.NoError(t, err) }) - s3agent := requireBoundOBC(ctx, t, k8sh, store, obc) + obc.RequireBound(ctx, t, k8sh, obc1) + s3agent := obc.NewS3Agent(ctx, t, k8sh, store, ns.Name, obc1.Name) - t.Run(fmt.Sprintf("notification %q is configured on bucket %q", notification.Name, obc.Spec.BucketName), func(t *testing.T) { - requireBucketNotification(ctx, t, s3agent, obc.Spec.BucketName, notification.Name, true) + t.Run(fmt.Sprintf("notification %q is configured on bucket %q", notification.Name, obc1.Spec.BucketName), func(t *testing.T) { + requireBucketNotification(ctx, t, s3agent, obc1.Spec.BucketName, notification.Name, true) }) t.Run(fmt.Sprintf("put object %q delivers a notification", key1), func(t *testing.T) { - _, err := s3agent.PutObjectInBucket(ctx, obc.Spec.BucketName, body, key1, contentType) + _, err := s3agent.PutObjectInBucket(ctx, obc1.Spec.BucketName, body, key1, contentType) require.NoError(t, err) requireNotificationReceived(ctx, t, k8sh, ns.Name, sinkSel, putEvent, key1) @@ -160,7 +161,7 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst }) t.Run(fmt.Sprintf("delete object %q delivers a notification", key1), func(t *testing.T) { - _, err := s3agent.DeleteObjectInBucket(ctx, obc.Spec.BucketName, key1) + _, err := s3agent.DeleteObjectInBucket(ctx, obc1.Spec.BucketName, key1) require.NoError(t, err) requireNotificationReceived(ctx, t, k8sh, ns.Name, sinkSel, deleteEvent, key1) @@ -168,43 +169,43 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst }) t.Run("an unrelated OBC label change leaves the notification configured", func(t *testing.T) { - t.Run(fmt.Sprintf("add a non-notification label to obc %q", obc.Name), func(t *testing.T) { - updateOBCLabels(ctx, t, k8sh, ns.Name, obc.Name, func(l map[string]string) { + t.Run(fmt.Sprintf("add a non-notification label to obc %q", obc1.Name), func(t *testing.T) { + updateOBCLabels(ctx, t, k8sh, ns.Name, obc1.Name, func(l map[string]string) { l["test-label"] = "test-value" }) }) - requireBucketNotification(ctx, t, s3agent, obc.Spec.BucketName, notification.Name, true) + requireBucketNotification(ctx, t, s3agent, obc1.Spec.BucketName, notification.Name, true) - t.Run(fmt.Sprintf("remove the non-notification label from obc %q", obc.Name), func(t *testing.T) { - updateOBCLabels(ctx, t, k8sh, ns.Name, obc.Name, func(l map[string]string) { + t.Run(fmt.Sprintf("remove the non-notification label from obc %q", obc1.Name), func(t *testing.T) { + updateOBCLabels(ctx, t, k8sh, ns.Name, obc1.Name, func(l map[string]string) { delete(l, "test-label") }) }) - requireBucketNotification(ctx, t, s3agent, obc.Spec.BucketName, notification.Name, true) + requireBucketNotification(ctx, t, s3agent, obc1.Spec.BucketName, notification.Name, true) }) t.Run("removing the notification label removes the notification from the bucket", func(t *testing.T) { - t.Run(fmt.Sprintf("remove the notification label from obc %q", obc.Name), func(t *testing.T) { - updateOBCLabels(ctx, t, k8sh, ns.Name, obc.Name, func(l map[string]string) { + t.Run(fmt.Sprintf("remove the notification label from obc %q", obc1.Name), func(t *testing.T) { + updateOBCLabels(ctx, t, k8sh, ns.Name, obc1.Name, func(l map[string]string) { delete(l, notificationLabel(notification.Name)) }) }) - requireBucketNotification(ctx, t, s3agent, obc.Spec.BucketName, notification.Name, false) + requireBucketNotification(ctx, t, s3agent, obc1.Spec.BucketName, notification.Name, false) }) t.Run("an unsubscribed bucket delivers no notifications", func(t *testing.T) { t.Run(fmt.Sprintf("put object %q", key2), func(t *testing.T) { - _, err := s3agent.PutObjectInBucket(ctx, obc.Spec.BucketName, body, key2, contentType) + _, err := s3agent.PutObjectInBucket(ctx, obc1.Spec.BucketName, body, key2, contentType) require.NoError(t, err) assertNotificationNotReceived(ctx, t, k8sh, ns.Name, sinkSel, putEvent, key2) }) t.Run(fmt.Sprintf("delete object %q", key2), func(t *testing.T) { - _, err := s3agent.DeleteObjectInBucket(ctx, obc.Spec.BucketName, key2) + _, err := s3agent.DeleteObjectInBucket(ctx, obc1.Spec.BucketName, key2) require.NoError(t, err) assertNotificationNotReceived(ctx, t, k8sh, ns.Name, sinkSel, deleteEvent, key2) @@ -224,13 +225,13 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst require.NoError(t, err) }) - t.Run(fmt.Sprintf("add the notification label for %q to obc %q", addNotification.Name, obc.Name), func(t *testing.T) { - updateOBCLabels(ctx, t, k8sh, ns.Name, obc.Name, func(l map[string]string) { + t.Run(fmt.Sprintf("add the notification label for %q to obc %q", addNotification.Name, obc1.Name), func(t *testing.T) { + updateOBCLabels(ctx, t, k8sh, ns.Name, obc1.Name, func(l map[string]string) { l[notificationLabel(addNotification.Name)] = addNotification.Name }) }) - requireBucketNotification(ctx, t, s3agent, obc.Spec.BucketName, addNotification.Name, true) + requireBucketNotification(ctx, t, s3agent, obc1.Spec.BucketName, addNotification.Name, true) t.Run(fmt.Sprintf("delete CephBucketNotification %q", addNotification.Name), func(t *testing.T) { wait4.AssertDelete(ctx, t, bnClient, addNotification.Name, wait4.TimeoutShort) @@ -244,9 +245,10 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst t.Run("a notification created before its topic is configured once the topic exists", func(t *testing.T) { revTopic := httpTopic(defaultName+"-topic-rev", ns.Name, objectStore, sinkURI) revNotification := bucketNotification(defaultName+"-notification-rev", ns.Name, revTopic.Name) - revOBC := notificationOBC(defaultName+"-obc-rev", ns.Name, storageClass.Name, revNotification.Name) + obcRev := notificationOBC(defaultName+"-obc-rev", ns.Name, storageClass.Name, revNotification.Name) - revS3 := requireBoundOBC(ctx, t, k8sh, store, revOBC) + obc.RequireBound(ctx, t, k8sh, obcRev) + revS3 := obc.NewS3Agent(ctx, t, k8sh, store, ns.Name, obcRev.Name) t.Run(fmt.Sprintf("create CephBucketNotification %q before its topic", revNotification.Name), func(t *testing.T) { _, err := bnClient.Create(ctx, revNotification, metav1.CreateOptions{}) @@ -257,8 +259,8 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst wait4.RequireCreate(ctx, t, btClient, revTopic, wait4.BucketTopic, wait4.TimeoutMedium) }) - t.Run(fmt.Sprintf("notification %q is configured on bucket %q once its topic exists", revNotification.Name, revOBC.Spec.BucketName), func(t *testing.T) { - requireBucketNotification(ctx, t, revS3, revOBC.Spec.BucketName, revNotification.Name, true) + t.Run(fmt.Sprintf("notification %q is configured on bucket %q once its topic exists", revNotification.Name, obcRev.Spec.BucketName), func(t *testing.T) { + requireBucketNotification(ctx, t, revS3, obcRev.Spec.BucketName, revNotification.Name, true) }) t.Run(fmt.Sprintf("delete CephBucketNotification %q", revNotification.Name), func(t *testing.T) { @@ -269,13 +271,13 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst wait4.AssertDelete(ctx, t, btClient, revTopic.Name, wait4.TimeoutShort) }) - t.Run(fmt.Sprintf("delete obc %q", revOBC.Name), func(t *testing.T) { - deleteOBCAndBucket(ctx, t, k8sh, ns.Name, revOBC.Name) + t.Run(fmt.Sprintf("delete obc %q", obcRev.Name), func(t *testing.T) { + obc.DeleteAndWait(ctx, t, k8sh, ns.Name, obcRev.Name) }) }) - t.Run(fmt.Sprintf("delete obc %q", obc.Name), func(t *testing.T) { - deleteOBCAndBucket(ctx, t, k8sh, ns.Name, obc.Name) + t.Run(fmt.Sprintf("delete obc %q", obc1.Name), func(t *testing.T) { + obc.DeleteAndWait(ctx, t, k8sh, ns.Name, obc1.Name) }) t.Run(fmt.Sprintf("delete CephBucketNotification %q", notification.Name), func(t *testing.T) { @@ -306,40 +308,6 @@ func TestBucketNotification(t *testing.T, k8sh *utils.K8sHelper, store *sharedst }) } -// requireBoundOBC creates obc, waits for it and its backing ObjectBucket to bind, -// and returns an S3 client for the provisioned bucket. -func requireBoundOBC(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore, obc *bktv1alpha1.ObjectBucketClaim) *rgw.S3Agent { - t.Helper() - - obcClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc.Namespace) - obClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets() - - var agent *rgw.S3Agent - t.Run(fmt.Sprintf("create obc %q", obc.Name), func(t *testing.T) { - liveOBC := wait4.RequireCreate(ctx, t, obcClient, obc, wait4.OBCBound, wait4.TimeoutLong) - wait4.RequireCondition(ctx, t, obClient, liveOBC.Spec.ObjectBucketName, wait4.OBBound, wait4.TimeoutShort) - - agent = obcS3Agent(ctx, t, k8sh, store, obc.Namespace, obc.Name) - }) - return agent -} - -// deleteOBCAndBucket deletes the OBC and waits for its backing ObjectBucket (and -// thus the rgw bucket) to be garbage-collected by the provisioner. -func deleteOBCAndBucket(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, namespace, name string) { - t.Helper() - - obcClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace) - obClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets() - - liveOBC, err := obcClient.Get(ctx, name, metav1.GetOptions{}) - require.NoError(t, err) - obName := liveOBC.Spec.ObjectBucketName - - wait4.AssertDelete(ctx, t, obcClient, name, wait4.TimeoutShort) - wait4.AssertAbsent(ctx, t, obClient, obName, wait4.TimeoutShort) -} - // updateOBCLabels applies mutate to a live OBC's labels and updates it. func updateOBCLabels(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, namespace, name string, mutate func(map[string]string)) { t.Helper() @@ -402,26 +370,6 @@ func requireHTTPSink(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, n }) } -// obcS3Agent builds an S3 client for the bucket provisioned by obcName, using the -// credentials in the OBC's secret and the shared store's endpoint. -func obcS3Agent(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore, namespace, obcName string) *rgw.S3Agent { - t.Helper() - - secret, err := k8sh.Clientset.CoreV1().Secrets(namespace).Get(ctx, obcName, metav1.GetOptions{}) - require.NoError(t, err) - - endpoint, err := client.GetS3Endpoint(store.ObjectStore(), k8sh, store.TLSEnabled()) - require.NoError(t, err) - - agent, err := rgw.NewS3Agent( - string(secret.Data["AWS_ACCESS_KEY_ID"]), - string(secret.Data["AWS_SECRET_ACCESS_KEY"]), - endpoint, true, nil, store.TLSEnabled(), nil) - require.NoError(t, err) - - return agent -} - // requireBucketNotification waits until notificationName is (want=true) or is not // (want=false) configured on the rgw bucket, read via the S3 API. func requireBucketNotification(ctx context.Context, t *testing.T, agent *rgw.S3Agent, bucket, notificationName string, want bool) { diff --git a/tests/integration/object/topic/kafka/kafka.go b/tests/integration/object/topic/kafka/kafka.go index af16287cf..dba2c6883 100644 --- a/tests/integration/object/topic/kafka/kafka.go +++ b/tests/integration/object/topic/kafka/kafka.go @@ -33,6 +33,7 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/tests/framework/utils" "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/obc" "github.com/rook/rook/tests/integration/object/util/secrets" "github.com/rook/rook/tests/integration/object/util/sharedstore" "github.com/rook/rook/tests/integration/object/util/wait4" @@ -137,7 +138,7 @@ func TestBucketTopicKafka(t *testing.T, k8sh *utils.K8sHelper, store *sharedstor secret3 = kvSecret(ns.Name, defaultName+"-secret3", "user-name", "kafka-user3") secret4 = kvSecret(ns.Name, defaultName+"-secret4", "password", "kafka-pass4") - storageClass = fixture.StorageClass(defaultName, objectStore) + storageClass = obc.StorageClass(defaultName, objectStore) bt1 = &cephv1.CephBucketTopic{ ObjectMeta: metav1.ObjectMeta{ diff --git a/tests/integration/object/util/fixture/obc.go b/tests/integration/object/util/fixture/obc.go deleted file mode 100644 index 4f06bb348..000000000 --- a/tests/integration/object/util/fixture/obc.go +++ /dev/null @@ -1,39 +0,0 @@ -/* -Copyright 2026 The Rook Authors. All rights reserved. - -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 fixture - -import ( - storagev1 "k8s.io/api/storage/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" -) - -// StorageClass returns a StorageClass for the lib-bucket-provisioner bucket -// provisioner backed by objectStore, for use by ObjectBucketClaims. -func StorageClass(name string, objectStore *cephv1.CephObjectStore) *storagev1.StorageClass { - return &storagev1.StorageClass{ - ObjectMeta: metav1.ObjectMeta{ - Name: name, - }, - Provisioner: objectStore.Namespace + ".ceph.rook.io/bucket", - Parameters: map[string]string{ - "objectStoreName": objectStore.Name, - "objectStoreNamespace": objectStore.Namespace, - }, - } -} diff --git a/tests/integration/object/util/obc/obc.go b/tests/integration/object/util/obc/obc.go new file mode 100644 index 000000000..04ed3d6ba --- /dev/null +++ b/tests/integration/object/util/obc/obc.go @@ -0,0 +1,104 @@ +/* +Copyright 2026 The Rook Authors. All rights reserved. + +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 obc holds the ObjectBucketClaim test helpers: the provisioner +// StorageClass constructor, the bound/delete lifecycle waiters, and a per-OBC S3 +// client. +package obc + +import ( + "context" + "fmt" + "testing" + + bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1" + "github.com/stretchr/testify/require" + storagev1 "k8s.io/api/storage/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + rgw "github.com/rook/rook/pkg/operator/ceph/object" + "github.com/rook/rook/tests/framework/utils" + "github.com/rook/rook/tests/integration/object/util/client" + "github.com/rook/rook/tests/integration/object/util/sharedstore" + "github.com/rook/rook/tests/integration/object/util/wait4" +) + +// StorageClass returns a StorageClass for the lib-bucket-provisioner bucket +// provisioner backed by objectStore, for use by ObjectBucketClaims. +func StorageClass(name string, objectStore *cephv1.CephObjectStore) *storagev1.StorageClass { + return &storagev1.StorageClass{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + }, + Provisioner: objectStore.Namespace + ".ceph.rook.io/bucket", + Parameters: map[string]string{ + "objectStoreName": objectStore.Name, + "objectStoreNamespace": objectStore.Namespace, + }, + } +} + +// RequireBound creates obc and waits for it and its backing ObjectBucket to +// bind, returning the live OBC; it aborts the caller if binding fails. +func RequireBound(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, obc *bktv1alpha1.ObjectBucketClaim) *bktv1alpha1.ObjectBucketClaim { + t.Helper() + + obcClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(obc.Namespace) + obClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets() + + var live *bktv1alpha1.ObjectBucketClaim + if !t.Run(fmt.Sprintf("create obc %q", obc.Name), func(t *testing.T) { + live = wait4.RequireCreate(ctx, t, obcClient, obc, wait4.OBCBound, wait4.TimeoutLong) + wait4.RequireCondition(ctx, t, obClient, live.Spec.ObjectBucketName, wait4.OBBound, wait4.TimeoutShort) + }) { + t.FailNow() + } + return live +} + +// NewS3Agent builds an S3 client for the bucket provisioned by obcName, using the +// credentials in the OBC's secret and the store's endpoint. +func NewS3Agent(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore, namespace, obcName string) *rgw.S3Agent { + t.Helper() + + secret, err := k8sh.Clientset.CoreV1().Secrets(namespace).Get(ctx, obcName, metav1.GetOptions{}) + require.NoError(t, err) + + agent, err := client.NewS3Agent( + store.ObjectStore(), k8sh, store.TLSEnabled(), + string(secret.Data["AWS_ACCESS_KEY_ID"]), + string(secret.Data["AWS_SECRET_ACCESS_KEY"])) + require.NoError(t, err) + + return agent +} + +// DeleteAndWait deletes the OBC and waits for its backing ObjectBucket (and thus +// the rgw bucket) to be garbage-collected by the provisioner. +func DeleteAndWait(ctx context.Context, t *testing.T, k8sh *utils.K8sHelper, namespace, name string) { + t.Helper() + + obcClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(namespace) + obClient := k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBuckets() + + live, err := obcClient.Get(ctx, name, metav1.GetOptions{}) + require.NoError(t, err) + obName := live.Spec.ObjectBucketName + + wait4.AssertDelete(ctx, t, obcClient, name, wait4.TimeoutShort) + wait4.AssertAbsent(ctx, t, obClient, obName, wait4.TimeoutShort) +} From b4d63353682ff5659cc972f9dd86c3aca3daec1b Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Wed, 1 Jul 2026 14:42:39 -0700 Subject: [PATCH 3/4] test: convert the object bucket read/write suite to the shared store Add tests/integration/object/bucket/rw, the first slice carved from the legacy testObjectStoreOperations. It provisions an OBC on the shared store, writes, reads back, and deletes an object over S3, and checks the OBC stays Bound. It runs against the shared store in both the TLS and non-TLS passes. Drop the converted put/get and OBC-revert subtests from the legacy test; the user quota subtest now seeds both of the objects it needs. The remaining quota, policy, lifecycle, and dependents slices convert in later PRs. Signed-off-by: Joshua Hoblitt --- tests/integration/ceph_object_test.go | 28 ++---- tests/integration/object/README.md | 19 ++-- tests/integration/object/bucket/rw/rw.go | 110 +++++++++++++++++++++++ 3 files changed, 125 insertions(+), 32 deletions(-) create mode 100644 tests/integration/object/bucket/rw/rw.go diff --git a/tests/integration/ceph_object_test.go b/tests/integration/ceph_object_test.go index b646346ac..098c81d1f 100644 --- a/tests/integration/ceph_object_test.go +++ b/tests/integration/ceph_object_test.go @@ -45,6 +45,7 @@ import ( "github.com/rook/rook/tests/framework/installer" "github.com/rook/rook/tests/framework/utils" bucketowner "github.com/rook/rook/tests/integration/object/bucket/owner" + bucketrw "github.com/rook/rook/tests/integration/object/bucket/rw" "github.com/rook/rook/tests/integration/object/cosi" "github.com/rook/rook/tests/integration/object/notification" topickafka "github.com/rook/rook/tests/integration/object/topic/kafka" @@ -213,6 +214,7 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, install sharedObjectStore := sharedstore.Create(s.T(), k8sh, installer, tlsEnable, bucketowner.Namespace, + bucketrw.Namespace, userkeys.Namespace, topickafka.Namespace, useropmask.Namespace, @@ -223,6 +225,7 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, install defer sharedObjectStore.Destroy() bucketowner.TestObjectBucketClaimBucketOwner(s.T(), k8sh, sharedObjectStore) + bucketrw.TestObjectBucketClaimReadWrite(s.T(), k8sh, sharedObjectStore) userkeys.TestObjectStoreUserKeys(s.T(), k8sh, sharedObjectStore) topickafka.TestBucketTopicKafka(s.T(), k8sh, sharedObjectStore) useropmask.TestObjectStoreUserOpMask(s.T(), k8sh, sharedObjectStore) @@ -298,19 +301,10 @@ func testObjectStoreOperations(s *suite.Suite, helper *clients.TestClient, k8sh assert.Nil(t, err) logger.Infof("endpoint (%s) Accesskey (%s) secret (%s)", s3endpoint, s3AccessKey, s3SecretKey) - t.Run("put object", func(t *testing.T) { + t.Run("user quota enforcement", func(t *testing.T) { _, poErr := s3client.PutObjectInBucket(ctx, bucketname, ObjBody, ObjectKey1, contentType) assert.Nil(t, poErr) - }) - - t.Run("get object", func(t *testing.T) { - read, err := s3client.GetObjectInBucket(ctx, bucketname, ObjectKey1) - assert.Nil(t, err) - assert.Equal(t, ObjBody, read) - }) - - t.Run("user quota enforcement", func(t *testing.T) { - _, poErr := s3client.PutObjectInBucket(ctx, bucketname, ObjBody, ObjectKey2, contentType) + _, poErr = s3client.PutObjectInBucket(ctx, bucketname, ObjBody, ObjectKey2, contentType) assert.Nil(t, poErr) logger.Infof("Testing the max object limit") quotaEnforced := utils.Retry(30, 2*time.Second, "user quota enforced", func() bool { @@ -1007,18 +1001,6 @@ func testObjectStoreOperations(s *suite.Suite, helper *clients.TestClient, k8sh }) }) - t.Run("Regression check: OBC does not revert to Pending phase", func(t *testing.T) { - // A bug exists in older versions of lib-bucket-provisioner that will revert a bucket and claim - // back to "Pending" phase after being created and initially "Bound" by looping infinitely in - // the bucket provision/creation loop. Verify that the OBC is "Bound" and stays that way. - // The OBC reconcile loop runs again immediately b/c the OBC is modified to refer to its OB. - // Wait a short amount of time before checking just to be safe. - created := utils.Retry(15, 2*time.Second, "OBC is created", func() bool { - return helper.BucketClient.CheckOBC(obcName, "bound") - }) - assert.True(t, created) - }) - t.Run("delete CephObjectStore should be blocked by OBC bucket and CephObjectStoreUser", func(t *testing.T) { deleteObjectStore(t, k8sh, namespace, storeName) diff --git a/tests/integration/object/README.md b/tests/integration/object/README.md index bf329ebd7..1b97e3360 100644 --- a/tests/integration/object/README.md +++ b/tests/integration/object/README.md @@ -33,13 +33,14 @@ must not collide with std-lib package names (no `io`, no `http`). | test package | operator package | covers | |---|---|---| | `bucket/owner` | `object/bucket` | OBC `bucketOwner` handling | +| `bucket/rw` | `object/bucket` | OBC S3 read/write/delete + OBC-stays-Bound | | `cosi` | `object/cosi` | CephCOSIDriver + COSI bucket provisioning | | `notification` | `object/notification` | CephBucketNotification HTTP endpoint delivery | | `topic/kafka` | `object/topic` | CephBucketTopic kafka endpoints | | `user/caps` | `object/user` | user capabilities | | `user/keys` | `object/user` | explicit S3 key management | | `user/opmask` | `object/user` | user op_mask | -| reserved: `bucket/rw`, `bucket/quota`, `bucket/policy`, `bucket/lifecycle`, `tests/integration/object/lifecycle`, `tests/integration/object/dependents` | | future conversions | +| reserved: `bucket/quota`, `bucket/policy`, `bucket/lifecycle`, `tests/integration/object/lifecycle`, `tests/integration/object/dependents` | | future conversions | Shared utilities live under `util/`: @@ -51,13 +52,14 @@ Shared utilities live under `util/`: `BucketTopic`, `OBCBound`, ...). - `fixture` — create-with-`t.Cleanup` helpers for pure-cleanup resources (namespaces, StorageClasses). -- `obc` — pure constructors for the resources ObjectBucketClaim tests build - (the provisioner `StorageClass`; OBCs and per-OBC S3 clients as `bucket/*` - is converted). +- `obc` — ObjectBucketClaim helpers: the provisioner `StorageClass` + constructor, the create/bound and delete/absent lifecycle waiters, and a + per-OBC S3 client. - `secrets` — verification helpers for the Secret references object CRDs publish in their status, shared by more than one package. - `sharedstore` — the shared CephObjectStore fixture. -- `admin`, `sns`, `s3`, `tls` — client builders and TLS cert generation. +- `admin`, `sns`, `s3`, `tls` — client builders (rgw admin, SNS, and a generic + S3 agent) and TLS cert generation. ## Anatomy of a package @@ -206,10 +208,9 @@ verification; wire the dispatcher; delete the old code; retire | old test | target package(s) | still needs (build in that PR) | |---|---|---| -| `testObjectStoreOperations` S3 I/O + OBC-stays-Bound | `bucket/rw` | per-OBC S3 client (TLS-aware, from OBC secret) | -| `testObjectStoreOperations` user/bucket quotas | `bucket/quota` | per-OBC S3 client | -| `testObjectStoreOperations` bucket policy | `bucket/policy` | per-OBC S3 client | -| `testObjectStoreOperations` bucket lifecycle | `bucket/lifecycle` | per-OBC S3 client, move `lifecycleCmpOpts` out of the dispatcher | +| `testObjectStoreOperations` user/bucket quotas | `bucket/quota` | — | +| `testObjectStoreOperations` bucket policy | `bucket/policy` | — | +| `testObjectStoreOperations` bucket lifecycle | `bucket/lifecycle` | move `lifecycleCmpOpts` out of the dispatcher | | `testObjectStoreOperations` deletion-blocked-by-dependents | `tests/integration/object/dependents` | private-store fixture, store condition predicates in `ready` | | `createCephObjectStore`/`runObjectE2ETestLite`/deletion asserts + zone.json canary | `tests/integration/object/lifecycle` | store create/health/delete helpers, `Sharedstore.Installer()` accessor | | upgrade-suite object usage | stays in upgrade suite | switch to typed clients + `wait4`/`ready` | diff --git a/tests/integration/object/bucket/rw/rw.go b/tests/integration/object/bucket/rw/rw.go new file mode 100644 index 000000000..863ddef41 --- /dev/null +++ b/tests/integration/object/bucket/rw/rw.go @@ -0,0 +1,110 @@ +/* +Copyright 2026 The Rook Authors. All rights reserved. + +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 rw + +import ( + "fmt" + "testing" + + bktv1alpha1 "github.com/kube-object-storage/lib-bucket-provisioner/pkg/apis/objectbucket.io/v1alpha1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/rook/rook/tests/framework/utils" + "github.com/rook/rook/tests/integration/object/util/fixture" + "github.com/rook/rook/tests/integration/object/util/obc" + "github.com/rook/rook/tests/integration/object/util/sharedstore" + "github.com/rook/rook/tests/integration/object/util/wait4" +) + +const Namespace = "test-bucketrw" + +func TestObjectBucketClaimReadWrite(t *testing.T, k8sh *utils.K8sHelper, store *sharedstore.Sharedstore) { + var ( + defaultName = Namespace + objectStore = store.ObjectStore() + + ns = &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: defaultName, + }, + } + + storageClass = obc.StorageClass(defaultName, objectStore) + + obc1 = &bktv1alpha1.ObjectBucketClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: defaultName + "-obc1", + Namespace: ns.Name, + }, + Spec: bktv1alpha1.ObjectBucketClaimSpec{ + BucketName: defaultName + "-obc1", + StorageClassName: storageClass.Name, + }, + } + + body = "test bucket read/write payload" + contentType = "text/plain" + key = "obj1" + + obcClient = k8sh.BucketClientset.ObjectbucketV1alpha1().ObjectBucketClaims(ns.Name) + ) + + t.Run("ObjectBucketClaim read/write", func(t *testing.T) { + ctx := t.Context() + + fixture.RequireNamespace(t, k8sh, ns) + fixture.RequireStorageClass(t, k8sh, storageClass) + + obc.RequireBound(ctx, t, k8sh, obc1) + s3 := obc.NewS3Agent(ctx, t, k8sh, store, ns.Name, obc1.Name) + + t.Run(fmt.Sprintf("put object %q", key), func(t *testing.T) { + _, err := s3.PutObjectInBucket(ctx, obc1.Spec.BucketName, body, key, contentType) + require.NoError(t, err) + }) + + t.Run(fmt.Sprintf("get object %q round-trips", key), func(t *testing.T) { + read, err := s3.GetObjectInBucket(ctx, obc1.Spec.BucketName, key) + require.NoError(t, err) + assert.Equal(t, body, read) + }) + + t.Run(fmt.Sprintf("delete object %q", key), func(t *testing.T) { + _, err := s3.DeleteObjectInBucket(ctx, obc1.Spec.BucketName, key) + assert.NoError(t, err) + }) + + t.Run(fmt.Sprintf("obc %q stays Bound", obc1.Name), func(t *testing.T) { + // regression: older lib-bucket-provisioner versions looped in the + // provision loop and reverted a Bound OBC back to Pending + wait4.AssertCondition(ctx, t, obcClient, obc1.Name, wait4.OBCBound, wait4.TimeoutShort) + }) + + t.Run(fmt.Sprintf("delete obc %q", obc1.Name), func(t *testing.T) { + obc.DeleteAndWait(ctx, t, k8sh, ns.Name, obc1.Name) + }) + + t.Run(fmt.Sprintf("no obc(s) in ns %q", ns.Name), func(t *testing.T) { + list, err := obcClient.List(ctx, metav1.ListOptions{}) + require.NoError(t, err) + assert.Len(t, list.Items, 0) + }) + }) +} From 064e37ae8d6d67e2902d6541ae6a48e31470c073 Mon Sep 17 00:00:00 2001 From: Joshua Hoblitt Date: Wed, 1 Jul 2026 16:26:34 -0700 Subject: [PATCH 4/4] test: reconcile the object README with the merged util packages The README still described the util layout from before the package consolidation, with ready, admin, sns, s3, and tls as standalone packages and a code sample calling ready.ObjectStoreUser. The readiness predicates live in wait4 and the client builders in client; update the package list, the code sample, and the playbook references to match. Signed-off-by: Joshua Hoblitt --- tests/integration/object/README.md | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/tests/integration/object/README.md b/tests/integration/object/README.md index 1b97e3360..2b86c6764 100644 --- a/tests/integration/object/README.md +++ b/tests/integration/object/README.md @@ -47,9 +47,8 @@ Shared utilities live under `util/`: - `wait4` — waits for object-test state: watch-based `Assert/Require` × `Create/Delete/Condition/Absent` for k8s resources, `Assert/RequireEventually` polling for non-k8s state, `Assert/RequirePodLog` for a matching line in a - pod's log stream, and the shared timeout tiers. -- `ready` — readiness predicates for the wait4 helpers (`ObjectStoreUser`, - `BucketTopic`, `OBCBound`, ...). + pod's log stream, the readiness predicates (`ObjectStoreUser`, `BucketTopic`, + `OBCBound`, ...), and the shared timeout tiers. - `fixture` — create-with-`t.Cleanup` helpers for pure-cleanup resources (namespaces, StorageClasses). - `obc` — ObjectBucketClaim helpers: the provisioner `StorageClass` @@ -58,8 +57,7 @@ Shared utilities live under `util/`: - `secrets` — verification helpers for the Secret references object CRDs publish in their status, shared by more than one package. - `sharedstore` — the shared CephObjectStore fixture. -- `admin`, `sns`, `s3`, `tls` — client builders (rgw admin, SNS, and a generic - S3 agent) and TLS cert generation. +- `client` — rgw admin, SNS, and S3 client builders and TLS cert generation. ## Anatomy of a package @@ -85,7 +83,7 @@ func TestObjectStoreUserCaps(t *testing.T, k8sh *utils.K8sHelper, store *shareds fixture.RequireNamespace(t, k8sh, ns) t.Run(fmt.Sprintf("create CephObjectStoreUser %q", osu1.Name), func(t *testing.T) { - wait4.RequireCreate(ctx, t, osuClient, &osu1, ready.ObjectStoreUser, wait4.TimeoutLong) + wait4.RequireCreate(ctx, t, osuClient, &osu1, wait4.ObjectStoreUser, wait4.TimeoutLong) }) // ... ordered subtests: act, verify, act, verify ... @@ -123,7 +121,7 @@ Rules encoded in that shape: `RequireCreate` (create + wait ready), `AssertDelete`/`RequireDelete` (delete + wait gone), `RequireCondition`/`AssertCondition` (wait for a predicate on an existing resource), `AssertAbsent`/`RequireAbsent` (wait - for cascade deletion you did not issue). Predicates come from `util/ready` + for cascade deletion you did not issue). Predicates come from `wait4` or are inlined when test-specific. - State NOT visible to the Kubernetes API (rgw admin ops, S3, SNS) uses `wait4.AssertEventually`/`RequireEventually`, the assert/require-flavored @@ -211,6 +209,6 @@ verification; wire the dispatcher; delete the old code; retire | `testObjectStoreOperations` user/bucket quotas | `bucket/quota` | — | | `testObjectStoreOperations` bucket policy | `bucket/policy` | — | | `testObjectStoreOperations` bucket lifecycle | `bucket/lifecycle` | move `lifecycleCmpOpts` out of the dispatcher | -| `testObjectStoreOperations` deletion-blocked-by-dependents | `tests/integration/object/dependents` | private-store fixture, store condition predicates in `ready` | +| `testObjectStoreOperations` deletion-blocked-by-dependents | `tests/integration/object/dependents` | private-store fixture, store condition predicates in `wait4` | | `createCephObjectStore`/`runObjectE2ETestLite`/deletion asserts + zone.json canary | `tests/integration/object/lifecycle` | store create/health/delete helpers, `Sharedstore.Installer()` accessor | -| upgrade-suite object usage | stays in upgrade suite | switch to typed clients + `wait4`/`ready` | +| upgrade-suite object usage | stays in upgrade suite | switch to typed clients + `wait4` |