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..2b86c6764 100644 --- a/tests/integration/object/README.md +++ b/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` | 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/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) + }) + }) +} 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/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 +} 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) +}