GetTestBucketAndSpec returns a bucket to be used during a test. The returned teardownFn MUST be called once the test is done, which closes the bucket, readies it for a new test, and releases back into the pool. persistentBucket flag determines behaviour for walrus buckets only; Couchbase bucket beha
(t testing.TB, persistentBucket bool)
| 383 | // persistentBucket flag determines behaviour for walrus buckets only; Couchbase bucket |
| 384 | // behaviour is defined by the bucket pool readier/init. |
| 385 | func (tbp *TestBucketPool) getTestBucketAndSpec(t testing.TB, persistentBucket bool) (b Bucket, s BucketSpec, teardownFn func(context.Context)) { |
| 386 | |
| 387 | ctx := TestCtx(t) |
| 388 | |
| 389 | // Return a new Walrus bucket when tbp has not been initialized |
| 390 | if !tbp.integrationMode { |
| 391 | tbp.Logf(ctx, "Getting walrus test bucket - tbp.integrationMode is not set") |
| 392 | var walrusURL string |
| 393 | if persistentBucket { |
| 394 | dir := t.TempDir() |
| 395 | walrusURL = rosmarUriFromPath(dir) |
| 396 | } else { |
| 397 | walrusURL = kTestWalrusURL |
| 398 | } |
| 399 | return tbp.GetWalrusTestBucket(t, walrusURL) |
| 400 | } |
| 401 | |
| 402 | if tbp.useExistingBucket { |
| 403 | tbp.Logf(ctx, "Using predefined bucket") |
| 404 | return tbp.GetExistingBucket(t) |
| 405 | } |
| 406 | |
| 407 | if atomic.LoadUint32(&tbp.preservedBucketCount) >= uint32(cap(tbp.readyBucketPool)) { |
| 408 | tbp.Logf(ctx, |
| 409 | "No more buckets available for testing. All pooled buckets have been preserved by failing tests.") |
| 410 | t.Skipf("No more buckets available for testing. All pooled buckets have been preserved for failing tests.") |
| 411 | } |
| 412 | |
| 413 | tbp.Logf(ctx, "Attempting to get test bucket from pool") |
| 414 | waitingBucketStart := time.Now() |
| 415 | var bucket Bucket |
| 416 | select { |
| 417 | case bucket = <-tbp.readyBucketPool: |
| 418 | case <-time.After(waitForReadyBucketTimeout): |
| 419 | tbp.Logf(ctx, "Timed out after %s waiting for a bucket to become available.", waitForReadyBucketTimeout) |
| 420 | t.Fatalf("TEST: Timed out after %s waiting for a bucket to become available.", waitForReadyBucketTimeout) |
| 421 | } |
| 422 | atomic.AddInt64(&tbp.stats.TotalWaitingForReadyBucketNano, time.Since(waitingBucketStart).Nanoseconds()) |
| 423 | |
| 424 | ctx = bucketCtx(ctx, bucket) |
| 425 | tbp.Logf(ctx, "Got test bucket from pool") |
| 426 | tbp.markBucketOpened(t, bucket) |
| 427 | |
| 428 | atomic.AddInt32(&tbp.stats.NumBucketsOpened, 1) |
| 429 | bucketOpenStart := time.Now() |
| 430 | bucketClosed := &AtomicBool{} |
| 431 | bucketSpec := getTestBucketSpec(tbp.clusterSpec, tbpBucketName(bucket.GetName())) |
| 432 | return bucket, bucketSpec, func(ctx context.Context) { |
| 433 | if !bucketClosed.CompareAndSwap(false, true) { |
| 434 | tbp.Logf(ctx, "Bucket teardown was already called. Ignoring.") |
| 435 | return |
| 436 | } |
| 437 | |
| 438 | tbp.Logf(ctx, "Teardown called - closing bucket") |
| 439 | atomic.AddInt32(&tbp.stats.NumBucketsClosed, 1) |
| 440 | atomic.AddInt64(&tbp.stats.TotalInuseBucketNano, time.Since(bucketOpenStart).Nanoseconds()) |
| 441 | tbp.markBucketClosed(t, bucket) |
| 442 | bucket.Close(ctx) |
no test coverage detected