add bucket content verification

Author: Elbehery

bucket.go

@@ -739,6 +739,10 @@ func (b *Bucket) write() []byte {
 	return value
 }
 
+func (b *Bucket) Write() []byte {
+	return b.write()
+}
+
 // rebalance attempts to balance all nodes.
 func (b *Bucket) rebalance() {
 	for _, n := range b.nodes {

bucket_test.go

@@ -475,7 +475,7 @@ func TestBucket_MoveBucket2(t *testing.T) {
 		{
 			name:          "happy path",
 			srcBucketPath: []string{"x", "y", "z"},
-			dstBucketPath: []string{"x", "y"},
+			dstBucketPath: []string{"a", "b"},
 			isInlined:     false,
 			expErr:        nil,
 		},
@@ -484,10 +484,64 @@ func TestBucket_MoveBucket2(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(*testing.T) {
 			db := btesting.MustCreateDB(t)
-			if err := db.Update(func(tx *bolt.Tx) error {
 
+			// arrange
+			if err := db.Update(func(tx *bolt.Tx) error {
 				bk := createBucketIfNotExist(t, tx, tc.srcBucketPath...)
-				insertRandKeysValuesBucket(t, bk, rand.Int())
+				insertRandKeysValuesBucket(t, bk, rand.Intn(4096))
+				createBucketIfNotExist(t, tx, tc.dstBucketPath...)
+
+				return nil
+			}); err != nil {
+				t.Fatal(err)
+			}
+			db.MustCheck()
+
+			// act
+			var tmpFile string
+			if err := db.Update(func(tx *bolt.Tx) error {
+				srcBucket := retrieveParentBucket(t, tx, tc.srcBucketPath...)
+				dstBucket := retrieveChildBucket(t, tx, tc.dstBucketPath...)
+				bucketToMove := tc.srcBucketPath[len(tc.srcBucketPath)-1]
+
+				// dump bucketToMove to bbolt file for assertion
+				bk := srcBucket.Bucket([]byte(bucketToMove))
+				tmpFile = dumpBucketToFile(t, bk)
+
+				mErr := srcBucket.MoveBucket([]byte(bucketToMove), dstBucket)
+				require.Equal(t, tc.expErr, mErr)
+
+				return nil
+			}); err != nil {
+				t.Fatal(err)
+			}
+			db.MustCheck()
+
+			// assert
+			if err := db.View(func(tx *bolt.Tx) error {
+				srcBucket := retrieveParentBucket(t, tx, tc.srcBucketPath...)
+				bucketToMove := tc.srcBucketPath[len(tc.srcBucketPath)-1]
+
+				if bk := srcBucket.Bucket([]byte(bucketToMove)); bk != nil {
+					t.Fatalf("expected childBucket %v to be moved from srcBucket %v", bucketToMove, srcBucket)
+				}
+
+				dstBucket := retrieveChildBucket(t, tx, tc.dstBucketPath...)
+				if bk := dstBucket.Bucket([]byte(bucketToMove)); bk == nil {
+					t.Fatalf("expected childBucket %v to be child of dstBucket %v", bucketToMove, dstBucket)
+				}
+
+				childBucket := dstBucket.Bucket([]byte(bucketToMove))
+				if childBucket == nil {
+					t.Fatalf("expected subBucket %v to exist within dstBucket %v", bucketToMove, dstBucket)
+				}
+
+				bucketOnDisk, err := os.ReadFile(tmpFile)
+				if err != nil {
+					t.Fatalf("error reading tmp file %v", tmpFile)
+				}
+
+				require.Equal(t, bucketOnDisk, childBucket.Write())
 
 				return nil
 			}); err != nil {
@@ -507,16 +561,57 @@ func createBucketIfNotExist(t testing.TB, tx *bolt.Tx, paths ...string) *bolt.Bu
 
 	for _, key := range paths[1:] {
 		bk, err = bk.CreateBucketIfNotExists([]byte(key))
-		t.Fatalf("error creating bucket %v: %v", key, err)
+		if err != nil {
+			t.Fatalf("error creating bucket %v: %v", key, err)
+		}
 	}
 
 	return bk
 }
 
+func retrieveParentBucket(t testing.TB, tx *bolt.Tx, paths ...string) *bolt.Bucket {
+	paths = paths[:len(paths)-1]
+	return retrieveChildBucket(t, tx, paths...)
+}
+
+func retrieveChildBucket(t testing.TB, tx *bolt.Tx, paths ...string) *bolt.Bucket {
+	t.Helper()
+
+	var bk *bolt.Bucket = nil
+	for _, path := range paths {
+		if bk == nil {
+			bk = tx.Bucket([]byte(path))
+		} else {
+			bk = bk.Bucket([]byte(path))
+		}
+		if bk == nil {
+			t.Fatalf("error retrieving bucket %v within paths %v", path, strings.TrimSuffix(strings.Join(paths, "->"), "->"))
+		}
+	}
+	return bk
+}
+
+func dumpBucketToFile(t testing.TB, bk *bolt.Bucket) string {
+	tmpFile := tempfile()
+	f, err := os.Create(tmpFile)
+	if err != nil {
+		t.Fatalf("error creating tmp file %v: %v", tmpFile, err)
+	}
+	defer f.Close()
+
+	data := bk.Write()
+	_, err = f.Write(data)
+	if err != nil {
+		t.Fatalf("error writing bucket %v to file %v tmpFile", bk, tmpFile)
+	}
+
+	return tmpFile
+}
+
 func insertRandKeysValuesBucket(t testing.TB, bk *bolt.Bucket, n int) {
 
-	var min, max = 1, 1024
 	var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
+	var min, max = 1, 1024
 
 	for i := 0; i < n; i++ {
 		// generate rand key/value length
@@ -526,9 +621,12 @@ func insertRandKeysValuesBucket(t testing.TB, bk *bolt.Bucket, n int) {
 		keyData := make([]rune, keyLength)
 		valData := make([]rune, valLength)
 
-		for i := range keyData {
-			keyData[i] = letters[rand.Intn(len(letters))]
-			valData[i] = letters[rand.Intn(len(letters))]
+		for j := range keyData {
+			keyData[j] = letters[rand.Intn(len(letters))]
+		}
+
+		for j := range valData {
+			valData[j] = letters[rand.Intn(len(letters))]
 		}
 
 		pErr := bk.Put([]byte(string(keyData)), []byte(string(valData)))