Merge pull request #17879 from jhoblitt/convert-bucket-rw-to-sharedstore

test: convert the object bucket read/write suite to the shared store

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)
 

tests/integration/object/README.md

@@ -33,31 +33,31 @@ 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/`:
 
 - `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` — 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.
+- `client` — rgw admin, SNS, and S3 client builders and TLS cert generation.
 
 ## Anatomy of a package
 
@@ -83,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 ...
@@ -121,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
@@ -206,10 +206,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` deletion-blocked-by-dependents | `tests/integration/object/dependents` | private-store fixture, store condition predicates in `ready` |
+| `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 `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` |

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)

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)
+		})
+	})
+}

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) {

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{

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
+}

tests/integration/object/util/fixture/obc.go

@@ -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,
-		},
-	}
-}

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)
+}