From d83f9809f1a132e29fee469fe5b159ee45f772ad Mon Sep 17 00:00:00 2001
From: guy-har <guy.hardonag@treeverse.io>
Date: Sun, 14 Jul 2024 14:18:23 +0300
Subject: [PATCH] Fix lakectl upload from standard input (#7984)

---
 cmd/lakectl/cmd/fs_upload.go | 18 ++++++++++++------
 esti/lakectl_test.go         |  9 +++++++--
 2 files changed, 19 insertions(+), 8 deletions(-)

diff --git a/cmd/lakectl/cmd/fs_upload.go b/cmd/lakectl/cmd/fs_upload.go
index 73bb08784dc..6d4c95cce18 100644
--- a/cmd/lakectl/cmd/fs_upload.go
+++ b/cmd/lakectl/cmd/fs_upload.go
@@ -33,12 +33,7 @@ var fsUploadCmd = &cobra.Command{
 		ctx, stop := signal.NotifyContext(ctx, os.Interrupt, os.Kill)
 		defer stop()
 
-		stat, err := os.Stat(source)
-		if err != nil {
-			Die("failed to stat source", 1)
-		}
-
-		if !recursive || !stat.IsDir() { // Ignore recursive if source is a file and not a directory
+		if !recursive || isFileOrStdin(source) {
 			if strings.HasSuffix(remotePath, uri.PathSeparator) {
 				Die("target path is not a valid URI", 1)
 			}
@@ -85,6 +80,17 @@ var fsUploadCmd = &cobra.Command{
 	},
 }
 
+func isFileOrStdin(source string) bool {
+	if source == StdinFileName {
+		return true
+	}
+	stat, err := os.Stat(source)
+	if err != nil {
+		Die("failed to stat source", 1)
+	}
+	return !stat.IsDir()
+}
+
 func upload(ctx context.Context, client apigen.ClientWithResponsesInterface, sourcePathname string, destURI *uri.URI, contentType string, syncFlags local.SyncFlags) (*apigen.ObjectStats, error) {
 	fp := Must(OpenByPath(sourcePathname))
 	defer func() {
diff --git a/esti/lakectl_test.go b/esti/lakectl_test.go
index a70b0dd106b..6f869c5ca00 100644
--- a/esti/lakectl_test.go
+++ b/esti/lakectl_test.go
@@ -112,8 +112,13 @@ func TestLakectlPreSignUpload(t *testing.T) {
 	RunCmdAndVerifySuccessWithFile(t, Lakectl()+" log lakefs://"+repoName+"/"+mainBranch, false, "lakectl_log_initial", vars)
 
 	filePath := "ro_1k.1"
-	vars["FILE_PATH"] = filePath
-	RunCmdAndVerifySuccessWithFile(t, Lakectl()+" fs upload -s files/ro_1k lakefs://"+repoName+"/"+mainBranch+"/"+filePath+" --pre-sign", false, "lakectl_fs_upload", vars)
+	t.Run("upload from file", func(t *testing.T) {
+		vars["FILE_PATH"] = filePath
+		RunCmdAndVerifySuccessWithFile(t, Lakectl()+" fs upload -s files/ro_1k lakefs://"+repoName+"/"+mainBranch+"/"+filePath+" --pre-sign", false, "lakectl_fs_upload", vars)
+	})
+	t.Run("upload from stdin", func(t *testing.T) {
+		RunCmdAndVerifySuccessWithFile(t, "cat files/ro_1k | "+Lakectl()+" fs upload -s - lakefs://"+repoName+"/"+mainBranch+"/"+filePath+" --pre-sign", false, "lakectl_fs_upload", vars)
+	})
 }
 
 func TestLakectlCommit(t *testing.T) {