Added new tests for streaming writes flow

tools/integration_tests/local_file/edit_file_test.go

@@ -0,0 +1,70 @@
+// Copyright 2025 Google LLC
+//
+// 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.
+
+// Provides integration tests for write large files sequentially and randomly.
+
+package local_file_test
+
+import (
+	"os"
+	"path"
+
+	. "github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/client"
+	"github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/operations"
+	"github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/setup"
+	"github.com/stretchr/testify/require"
+)
+
+func (t *CommonLocalFileTestSuite) TestEditsToNewlyCreatedFile() {
+	testDirPath = setup.SetupTestDirectory(testDirName)
+	// Create a local file.
+	_, fh := CreateLocalFileInTestDir(ctx, storageClient, testDirPath, FileName1, t.T())
+	// Write some contents to file sequentially.
+	for i := 0; i < 3; i++ {
+		operations.WriteWithoutClose(fh, FileContents, t.T())
+	}
+	// Close the file and validate that the file is created on GCS.
+	expectedContent := FileContents + FileContents + FileContents
+	CloseFileAndValidateContentFromGCS(ctx, storageClient, fh, testDirName, FileName1, expectedContent, t.T())
+
+	// Perform edit
+	fhNew := operations.OpenFile(path.Join(testDirPath, FileName1), t.T())
+	newContent := "newContent"
+	_, err := fhNew.WriteAt([]byte(newContent), 0)
+
+	require.Nil(t.T(), err)
+	CloseFileAndValidateContentFromGCS(ctx, storageClient, fhNew, testDirName, FileName1, newContent+FileContents+FileContents, t.T())
+}
+
+func (t *CommonLocalFileTestSuite) TestAppendsToNewlyCreatedFile() {
+	testDirPath = setup.SetupTestDirectory(testDirName)
+	// Create a local file.
+	_, fh := CreateLocalFileInTestDir(ctx, storageClient, testDirPath, FileName1, t.T())
+	// Write some contents to file sequentially.
+	for i := 0; i < 3; i++ {
+		operations.WriteWithoutClose(fh, FileContents, t.T())
+	}
+	// Close the file and validate that the file is created on GCS.
+	expectedContent := FileContents + FileContents + FileContents
+	CloseFileAndValidateContentFromGCS(ctx, storageClient, fh, testDirName, FileName1, expectedContent, t.T())
+
+	// Append to the file.
+	fhNew, err := os.OpenFile(path.Join(testDirPath, FileName1), os.O_RDWR|os.O_APPEND, operations.FilePermission_0777)
+	require.Nil(t.T(), err)
+	appendedContent := "appendedContent"
+	_, err = fhNew.Write([]byte(appendedContent))
+
+	require.Nil(t.T(), err)
+	CloseFileAndValidateContentFromGCS(ctx, storageClient, fhNew, testDirName, FileName1, expectedContent+appendedContent, t.T())
+}

tools/integration_tests/streaming_writes/read_file_test.go

@@ -0,0 +1,58 @@
+// Copyright 2025 Google LLC
+//
+// 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.
+
+// Provides integration tests for write large files sequentially and randomly.
+
+package streaming_writes
+
+import (
+	"path"
+
+	. "github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/client"
+	"github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/operations"
+	"github.com/stretchr/testify/require"
+)
+
+func (t *defaultMountCommonTest) TestReadBeforeFileIsFlushed() {
+	testContent := "testContent"
+	// Write data to file.
+	operations.WriteAt(testContent, 0, t.f1, t.T())
+
+	// Try to read the file.
+	_, err := t.f1.Seek(0, 0)
+	require.NoError(t.T(), err)
+	buf := make([]byte, 10)
+	_, err = t.f1.Read(buf)
+
+	require.Error(t.T(), err, "input/output error")
+	// Validate if correct content is uploaded to GCS after read error.
+	CloseFileAndValidateContentFromGCS(ctx, storageClient, t.f1, testDirName, t.fileName, testContent, t.T())
+}
+
+func (t *defaultMountCommonTest) TestReadAfterFlush() {
+	testContent := "testContent"
+	// Write data to file and flush.
+	operations.WriteAt(testContent, 0, t.f1, t.T())
+	CloseFileAndValidateContentFromGCS(ctx, storageClient, t.f1, testDirName, t.fileName, testContent, t.T())
+
+	// Perform read and validate the contents.
+	var err error
+	t.f1, err = operations.OpenFileAsReadonly(path.Join(testDirPath, t.fileName))
+	require.NoError(t.T(), err)
+	buf := make([]byte, len(testContent))
+	_, err = t.f1.Read(buf)
+
+	require.NoError(t.T(), err)
+	require.Equal(t.T(), string(buf), testContent)
+}

tools/integration_tests/util/operations/file_operations.go

@@ -230,26 +230,32 @@ func ReadFileSequentially(filePath string, chunkSize int64) (content []byte, err
 
 // Write data of chunkSize in file at given offset.
 func WriteChunkOfRandomBytesToFile(file *os.File, chunkSize int, offset int64) error {
+	return WriteChunkOfRandomBytesToFiles([]*os.File{file}, chunkSize, offset)
+}
+
+func WriteChunkOfRandomBytesToFiles(files []*os.File, chunkSize int, offset int64) error {
 	// Generate random data of chunk size.
 	chunk := make([]byte, chunkSize)
 	_, err := rand.Read(chunk)
 	if err != nil {
 		return fmt.Errorf("error while generating random string: %v", err)
 	}
 
-	// Write data in the file.
-	n, err := file.WriteAt(chunk, offset)
-	if err != nil {
-		return fmt.Errorf("error in writing randomly in file: %v", err)
-	}
+	for _, file := range files {
+		// Write data in the file.
+		n, err := file.WriteAt(chunk, offset)
+		if err != nil {
+			return fmt.Errorf("error in writing randomly in file: %s, %v", file.Name(), err)
+		}
 
-	if n != chunkSize {
-		return fmt.Errorf("incorrect number of bytes written in the file actual %d, expected %d", n, chunkSize)
-	}
+		if n != chunkSize {
+			return fmt.Errorf("incorrect number of bytes written in the file %s actual %d, expected %d", file.Name(), n, chunkSize)
+		}
 
-	err = file.Sync()
-	if err != nil {
-		return fmt.Errorf("error in syncing file: %v", err)
+		err = file.Sync()
+		if err != nil {
+			return fmt.Errorf("error in syncing file: %v", err)
+		}
 	}
 
 	return nil

tools/integration_tests/write_large_files/concurrent_write_to_same_file_test.go

@@ -0,0 +1,100 @@
+// Copyright 2025 Google LLC
+//
+// 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.
+
+// Provides integration tests for write large files sequentially and randomly.
+
+package write_large_files
+
+import (
+	"os"
+	"path"
+	"syscall"
+	"testing"
+
+	"github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/operations"
+	"github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/setup"
+	"golang.org/x/sync/errgroup"
+)
+
+func TestWriteToSameFileConcurrently(t *testing.T) {
+	seqWriteDir := setup.SetupTestDirectory(DirForSeqWrite)
+	mountedFilePath := path.Join(seqWriteDir, "50mb"+setup.GenerateRandomString(5)+".txt")
+	localFilePath := path.Join(TmpDir, "50mbLocal"+setup.GenerateRandomString(5)+".txt")
+	localFile := operations.CreateFile(localFilePath, setup.FilePermission_0600, t)
+
+	// Clean up.
+	defer operations.RemoveDir(seqWriteDir)
+	defer operations.RemoveFile(localFilePath)
+
+	var eG errgroup.Group
+	concurrentWriterCount := 5
+	chunkSize := 50 * OneMiB / concurrentWriterCount
+
+	// We will have x numbers of concurrent threads trying to write from the same file.
+	// Every thread will start at offset = thread_index * (fileSize/thread_count).
+	for i := 0; i < concurrentWriterCount; i++ {
+		offset := i * chunkSize
+
+		eG.Go(func() error {
+			return writeToFileSequentially(localFilePath, mountedFilePath, offset, offset+chunkSize, t)
+		})
+	}
+
+	// Wait on threads to end.
+	err := eG.Wait()
+	if err != nil {
+		t.Errorf("writing failed")
+	}
+
+	// Close the local file since the below method will open the file again.
+	operations.CloseFile(localFile)
+
+	identical, err := operations.AreFilesIdentical(mountedFilePath, localFilePath)
+	if !identical {
+		t.Fatalf("Comparision failed: %v", err)
+	}
+}
+
+func writeToFileSequentially(localFilePath string, mountedFilePath string, startOffset int, endOffset int, t *testing.T) (err error) {
+	mountedFile, err := os.OpenFile(mountedFilePath, os.O_RDWR|syscall.O_DIRECT|os.O_CREATE, setup.FilePermission_0600)
+	if err != nil {
+		t.Fatalf("Error in opening file: %v", err)
+	}
+
+	localFile, err := os.OpenFile(localFilePath, os.O_RDWR|syscall.O_DIRECT|os.O_CREATE, setup.FilePermission_0600)
+	if err != nil {
+		t.Fatalf("Error in opening file: %v", err)
+	}
+
+	filesToWrite := []*os.File{localFile, mountedFile}
+
+	// Closing file at the end.
+	defer operations.CloseFile(mountedFile)
+	defer operations.CloseFile(localFile)
+
+	var chunkSize = 5 * OneMiB
+	for startOffset < endOffset {
+		if (endOffset - startOffset) < chunkSize {
+			chunkSize = endOffset - startOffset
+		}
+
+		err := operations.WriteChunkOfRandomBytesToFiles(filesToWrite, chunkSize, int64(startOffset))
+		if err != nil {
+			t.Fatalf("Error in writing chunk: %v", err)
+		}
+
+		startOffset = startOffset + chunkSize
+	}
+	return
+}

tools/integration_tests/write_large_files/write_large_files_test.go

@@ -54,8 +54,12 @@ func compareFileFromGCSBucketAndMntDir(gcsFile, mntDirFile, localFilePathToDownl
 func TestMain(m *testing.M) {
 	setup.ParseSetUpFlags()
 
-	// TODO: remove max-blocks-per-file after the default values are set.
-	flags := [][]string{{"--enable-streaming-writes=true", "--write-max-blocks-per-file=2"}}
+	// write-global-max-blocks=2 is for checking multiple file writes in parallel.
+	// concurrent_write_files_test.go- we are writing 3 files in parallel.
+	// with this config, we are giving 2 blocks to 2 files and 1 block to other file.
+	flags := [][]string{
+		{"--enable-streaming-writes=false"},
+		{"--enable-streaming-writes=true", "--write-max-blocks-per-file=2", "--write-global-max-blocks=2"}}
 
 	setup.ExitWithFailureIfBothTestBucketAndMountedDirectoryFlagsAreNotSet()