Fix existing local file integration tests for streaming writes flow.

tools/integration_tests/local_file/local_file_test.go

@@ -98,7 +98,10 @@ func TestMain(m *testing.M) {
 	// Not setting config file explicitly with 'create-empty-file: false' as it is default.
 	flagsSet := [][]string{
 		{"--implicit-dirs=true", "--rename-dir-limit=3"},
-		{"--implicit-dirs=false", "--rename-dir-limit=3"}}
+		{"--implicit-dirs=false", "--rename-dir-limit=3"},
+		{"--enable-streaming-writes=true", "--write-block-size-mb=2", "--write-max-blocks-per-file=2"},
+		{"--enable-streaming-writes=true", "--write-block-size-mb=2", "--write-max-blocks-per-file=2", "create-empty-file: true"},
+	}
 
 	if hnsFlagSet, err := setup.AddHNSFlagForHierarchicalBucket(ctx, storageClient); err == nil {
 		flagsSet = append(flagsSet, hnsFlagSet)

tools/integration_tests/local_file/read_file_test.go

@@ -32,12 +32,15 @@ func TestReadLocalFile(t *testing.T) {
 	WritingToLocalFileShouldNotWriteToGCS(ctx, storageClient, fh, testDirName, FileName1, t)
 	WritingToLocalFileShouldNotWriteToGCS(ctx, storageClient, fh, testDirName, FileName1, t)
 
-	// Read the local file contents.
-	buf := make([]byte, len(content))
-	n, err := fh.ReadAt(buf, 0)
-	if err != nil || len(content) != n || content != string(buf) {
-		t.Fatalf("Read file operation failed on local file: %v "+
-			"Expected content: %s, Got Content: %s", err, content, string(buf))
+	// mounts with streaming writes disabled support read operation.
+	if !setup.StreamingWritesEnabled() {
+		// Read the local file contents.
+		buf := make([]byte, len(content))
+		n, err := fh.ReadAt(buf, 0)
+		if err != nil || len(content) != n || content != string(buf) {
+			t.Fatalf("Read file operation failed on local file: %v "+
+				"Expected content: %s, Got Content: %s", err, content, string(buf))
+		}
 	}
 
 	// Close the file and validate that the file is created on GCS.

tools/integration_tests/local_file/stat_file_test.go

@@ -23,6 +23,7 @@ import (
 	. "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 TestStatOnLocalFile(t *testing.T) {
@@ -57,7 +58,7 @@ func TestStatOnLocalFileWithConflictingFileNameSuffix(t *testing.T) {
 		FileName1, "", t)
 }
 
-func TestTruncateLocalFile(t *testing.T) {
+func TestTruncateLocalFileToSmallerSize(t *testing.T) {
 	testDirPath = setup.SetupTestDirectory(testDirName)
 	// Create a local file.
 	filePath, fh := CreateLocalFileInTestDir(ctx, storageClient, testDirPath, FileName1, t)
@@ -67,17 +68,53 @@ func TestTruncateLocalFile(t *testing.T) {
 	// Stat the file to validate if new contents are written.
 	operations.VerifyStatFile(filePath, SizeOfFileContents, FilePerms, t)
 
-	// Truncate the file to update the file size.
-	err := os.Truncate(filePath, SizeTruncate)
-	if err != nil {
-		t.Fatalf("os.Truncate err: %v", err)
+	// Truncate the file to update file size to smaller file size.
+	err := os.Truncate(filePath, SmallerSizeTruncate)
+	var expectTruncatedSize int64 = SizeOfFileContents
+	var expectedContent = FileContents
+	if setup.StreamingWritesEnabled() {
+		// Mounts with streaming writes do not supporting truncating files to smaller.
+		require.Error(t, err)
+	} else {
+		if err != nil {
+			t.Fatalf("os.Truncate err: %v", err)
+		}
+		expectTruncatedSize = SmallerSizeTruncate
+		expectedContent = FileContents[:SmallerSizeTruncate]
 	}
+
 	ValidateObjectNotFoundErrOnGCS(ctx, storageClient, testDirName, FileName1, t)
 
 	// Stat the file to validate if file is truncated correctly.
-	operations.VerifyStatFile(filePath, SizeTruncate, FilePerms, t)
+	operations.VerifyStatFile(filePath, expectTruncatedSize, FilePerms, t)
 
 	// Close the file and validate that the file is created on GCS.
 	CloseFileAndValidateContentFromGCS(ctx, storageClient, fh, testDirName,
-		FileName1, "testS", t)
+		FileName1, expectedContent, t)
+}
+
+func TestTruncateLocalFileToBiggerSize(t *testing.T) {
+	testDirPath = setup.SetupTestDirectory(testDirName)
+	// Create a local file.
+	filePath, fh := CreateLocalFileInTestDir(ctx, storageClient, testDirPath, FileName1, t)
+	// Writing contents to local file .
+	WritingToLocalFileShouldNotWriteToGCS(ctx, storageClient, fh, testDirName, FileName1, t)
+
+	// Stat the file to validate if new contents are written.
+	operations.VerifyStatFile(filePath, SizeOfFileContents, FilePerms, t)
+
+	// Truncate the file to update file size to bigger file size.
+	err := os.Truncate(filePath, BiggerSizeTruncate)
+	if err != nil {
+		t.Fatalf("os.Truncate err: %v", err)
+	}
+
+	ValidateObjectNotFoundErrOnGCS(ctx, storageClient, testDirName, FileName1, t)
+
+	// Stat the file to validate if file is truncated correctly.
+	operations.VerifyStatFile(filePath, BiggerSizeTruncate, FilePerms, t)
+
+	// Close file and validate that file of expected size is created on GCS.
+	CloseFileAndValidateSizeFromGCS(ctx, storageClient, fh, testDirName,
+		FileName1, BiggerSizeTruncate, t)
 }

tools/integration_tests/local_file/sym_link_test.go

@@ -26,7 +26,7 @@ import (
 	"github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/setup"
 )
 
-func createAndVerifySymLink(t *testing.T) (filePath, symlink string, fh *os.File) {
+func createAndVerifySymLink(streamingWritesEnabled bool, t *testing.T) (filePath, symlink string, fh *os.File) {
 	testDirPath = setup.SetupTestDirectory(testDirName)
 	// Create a local file.
 	filePath, fh = CreateLocalFileInTestDir(ctx, storageClient, testDirPath, FileName1, t)
@@ -38,18 +38,23 @@ func createAndVerifySymLink(t *testing.T) (filePath, symlink string, fh *os.File
 
 	// Read the link.
 	operations.VerifyReadLink(filePath, symlink, t)
+	if streamingWritesEnabled {
+		// Mounts with streaming writes do not support reading files.
+		return
+	}
 	operations.VerifyReadFile(symlink, FileContents, t)
 	return
 }
 
 func TestCreateSymlinkForLocalFile(t *testing.T) {
-	_, _, fh := createAndVerifySymLink(t)
+	_, _, fh := createAndVerifySymLink(setup.StreamingWritesEnabled(), t)
+	// Close the file and validate that the file is created on GCS.
 	CloseFileAndValidateContentFromGCS(ctx, storageClient, fh, testDirName,
 		FileName1, FileContents, t)
 }
 
 func TestReadSymlinkForDeletedLocalFile(t *testing.T) {
-	filePath, symlink, fh := createAndVerifySymLink(t)
+	filePath, symlink, fh := createAndVerifySymLink(setup.StreamingWritesEnabled(), t)
 	// Remove filePath and then close the fileHandle to avoid syncing to GCS.
 	operations.RemoveFile(filePath)
 	operations.CloseFileShouldNotThrowError(fh, t)

tools/integration_tests/local_file/write_file_test.go

@@ -46,9 +46,16 @@ func TestRandomWritesToLocalFile(t *testing.T) {
 
 	// Write some contents to file randomly.
 	operations.WriteAt("string1", 0, fh, t)
+	ValidateObjectNotFoundErrOnGCS(ctx, storageClient, testDirName, FileName1, t)
 	operations.WriteAt("string2", 2, fh, t)
+	if setup.StreamingWritesEnabled() {
+		// First out of order write ensures the existing sequentially written data is uploaded
+		// to GCS when streaming writes are enabled.
+		ValidateObjectContentsFromGCS(ctx, storageClient, testDirName, FileName1, "string1", t)
+	} else {
+		ValidateObjectNotFoundErrOnGCS(ctx, storageClient, testDirName, FileName1, t)
+	}
 	operations.WriteAt("string3", 3, fh, t)
-	ValidateObjectNotFoundErrOnGCS(ctx, storageClient, testDirName, FileName1, t)
 
 	// Close the file and validate that the file is created on GCS.
 	CloseFileAndValidateContentFromGCS(ctx, storageClient, fh, testDirName,

tools/integration_tests/util/client/gcs_helper.go

@@ -42,7 +42,8 @@ const (
 	GCSFileContent         = "GCSteststring"
 	GCSFileSize            = 13
 	FilePerms              = 0644
-	SizeTruncate           = 5
+	SmallerSizeTruncate    = 5
+	BiggerSizeTruncate     = 15
 	NewFileName            = "newName"
 	NewDirName             = "newDirName"
 )
@@ -79,6 +80,18 @@ func ValidateObjectContentsFromGCS(ctx context.Context, storageClient *storage.C
 	}
 }
 
+func ValidateObjectSizeFromGCS(ctx context.Context, storageClient *storage.Client,
+	testDirName string, fileName string, expectedSize int64, t *testing.T) {
+	gotContent, err := ReadObjectFromGCS(ctx, storageClient, path.Join(testDirName, fileName))
+	if err != nil {
+		t.Fatalf("Error while reading file from GCS, Err: %v", err)
+	}
+
+	if int64(len(gotContent)) != expectedSize {
+		t.Fatalf("GCS file %s size mismatch. Got file size: %d, Expected file size: %d ", fileName, len(gotContent), expectedSize)
+	}
+}
+
 func ValidateObjectChunkFromGCS(ctx context.Context, storageClient *storage.Client,
 	testDirName string, fileName string, offset, size int64, expectedContent string,
 	t *testing.T) {
@@ -100,6 +113,12 @@ func CloseFileAndValidateContentFromGCS(ctx context.Context, storageClient *stor
 	ValidateObjectContentsFromGCS(ctx, storageClient, testDirName, fileName, content, t)
 }
 
+func CloseFileAndValidateSizeFromGCS(ctx context.Context, storageClient *storage.Client,
+	fh *os.File, testDirName, fileName string, size int64, t *testing.T) {
+	operations.CloseFileShouldNotThrowError(fh, t)
+	ValidateObjectSizeFromGCS(ctx, storageClient, testDirName, fileName, size, t)
+}
+
 func CreateLocalFileInTestDir(ctx context.Context, storageClient *storage.Client,
 	testDirPath, fileName string, t *testing.T) (string, *os.File) {
 	filePath := path.Join(testDirPath, fileName)

tools/integration_tests/util/mounting/mounting.go

@@ -29,7 +29,8 @@ func MountGcsfuse(binaryFile string, flags []string) error {
 		binaryFile,
 		flags...,
 	)
-
+	// Sets to true iff current mount operation is using streaming writes
+	setup.SetStreamingWritesEnabled(flags)
 	// Adding mount command in LogFile
 	file, err := os.OpenFile(setup.LogFile(), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
 	if err != nil {

tools/integration_tests/util/setup/setup.go

@@ -48,16 +48,18 @@ const (
 	DirPermission_0755  = 0755
 	Charset             = "abcdefghijklmnopqrstuvwxyz0123456789"
 	PathEnvVariable     = "PATH"
+	StreamingWritesFlag = "--enable-streaming-writes=true"
 )
 
 var (
-	binFile              string
-	logFile              string
-	testDir              string
-	mntDir               string
-	sbinFile             string
-	onlyDirMounted       string
-	dynamicBucketMounted string
+	binFile                string
+	logFile                string
+	testDir                string
+	mntDir                 string
+	sbinFile               string
+	onlyDirMounted         string
+	dynamicBucketMounted   string
+	streamingWritesEnabled bool
 )
 
 // Run the shell script to prepare the testData in the specified bucket.
@@ -103,6 +105,19 @@ func LogFile() string {
 	return logFile
 }
 
+func SetStreamingWritesEnabled(flags []string) {
+	streamingWritesEnabled = false
+	for _, flag := range flags {
+		if flag == StreamingWritesFlag {
+			streamingWritesEnabled = true
+		}
+	}
+}
+
+func StreamingWritesEnabled() bool {
+	return streamingWritesEnabled
+}
+
 func BinFile() string {
 	return binFile
 }