Add fs mkdirs command for dbfs (#432)

## Changes
TSIA

## Tests
Integration tests

cmd/fs/mkdir.go

@@ -0,0 +1,39 @@
+package fs
+
+import (
+	"github.com/databricks/cli/cmd/root"
+	"github.com/databricks/cli/libs/filer"
+	"github.com/spf13/cobra"
+)
+
+var mkdirCmd = &cobra.Command{
+	Use: "mkdir DIR_PATH",
+	// Alias `mkdirs` for this command exists for legacy purposes. This command
+	// is called databricks fs mkdirs in our legacy CLI: https://github.com/databricks/databricks-cli
+	Aliases: []string{"mkdirs"},
+	Short:   "Make directories",
+	Long:    `Mkdir will create directories along the path to the argument directory.`,
+	Args:    cobra.ExactArgs(1),
+	PreRunE: root.MustWorkspaceClient,
+
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx := cmd.Context()
+		w := root.WorkspaceClient(ctx)
+
+		path, err := resolveDbfsPath(args[0])
+		if err != nil {
+			return err
+		}
+
+		f, err := filer.NewDbfsClient(w, "/")
+		if err != nil {
+			return err
+		}
+
+		return f.Mkdir(ctx, path)
+	},
+}
+
+func init() {
+	fsCmd.AddCommand(mkdirCmd)
+}

internal/mkdir_test.go

@@ -0,0 +1,114 @@
+package internal
+
+import (
+	"context"
+	"path"
+	"strings"
+	"testing"
+
+	"github.com/databricks/cli/libs/filer"
+	"github.com/databricks/databricks-sdk-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TesFsMkdirCreatesDirectory(t *testing.T) {
+	t.Log(GetEnvOrSkipTest(t, "CLOUD_ENV"))
+
+	ctx := context.Background()
+	w, err := databricks.NewWorkspaceClient()
+	require.NoError(t, err)
+
+	tmpDir := temporaryDbfsDir(t, w)
+
+	f, err := filer.NewDbfsClient(w, tmpDir)
+	require.NoError(t, err)
+
+	// create directory "a"
+	stdout, stderr := RequireSuccessfulRun(t, "fs", "mkdir", "dbfs:"+path.Join(tmpDir, "a"))
+	assert.Equal(t, "", stderr.String())
+	assert.Equal(t, "", stdout.String())
+
+	// assert directory "a" is created
+	info, err := f.Stat(ctx, "a")
+	require.NoError(t, err)
+	assert.Equal(t, "a", info.Name())
+	assert.Equal(t, true, info.IsDir())
+}
+
+func TestFsMkdirCreatesMultipleDirectories(t *testing.T) {
+	t.Log(GetEnvOrSkipTest(t, "CLOUD_ENV"))
+
+	ctx := context.Background()
+	w, err := databricks.NewWorkspaceClient()
+	require.NoError(t, err)
+
+	tmpDir := temporaryDbfsDir(t, w)
+
+	f, err := filer.NewDbfsClient(w, tmpDir)
+	require.NoError(t, err)
+
+	// create directory /a/b/c
+	stdout, stderr := RequireSuccessfulRun(t, "fs", "mkdir", "dbfs:"+path.Join(tmpDir, "a", "b", "c"))
+	assert.Equal(t, "", stderr.String())
+	assert.Equal(t, "", stdout.String())
+
+	// assert directory "a" is created
+	infoA, err := f.Stat(ctx, "a")
+	require.NoError(t, err)
+	assert.Equal(t, "a", infoA.Name())
+	assert.Equal(t, true, infoA.IsDir())
+
+	// assert directory "b" is created
+	infoB, err := f.Stat(ctx, "a/b")
+	require.NoError(t, err)
+	assert.Equal(t, "b", infoB.Name())
+	assert.Equal(t, true, infoB.IsDir())
+
+	// assert directory "c" is created
+	infoC, err := f.Stat(ctx, "a/b/c")
+	require.NoError(t, err)
+	assert.Equal(t, "c", infoC.Name())
+	assert.Equal(t, true, infoC.IsDir())
+}
+
+func TestFsMkdirWhenDirectoryAlreadyExists(t *testing.T) {
+	t.Log(GetEnvOrSkipTest(t, "CLOUD_ENV"))
+
+	ctx := context.Background()
+	w, err := databricks.NewWorkspaceClient()
+	require.NoError(t, err)
+
+	tmpDir := temporaryDbfsDir(t, w)
+
+	// create directory "a"
+	f, err := filer.NewDbfsClient(w, tmpDir)
+	require.NoError(t, err)
+	err = f.Mkdir(ctx, "a")
+	require.NoError(t, err)
+
+	// assert run is successful without any errors
+	stdout, stderr := RequireSuccessfulRun(t, "fs", "mkdir", "dbfs:"+path.Join(tmpDir, "a"))
+	assert.Equal(t, "", stderr.String())
+	assert.Equal(t, "", stdout.String())
+}
+
+func TestFsMkdirWhenFileExistsAtPath(t *testing.T) {
+	t.Log(GetEnvOrSkipTest(t, "CLOUD_ENV"))
+
+	ctx := context.Background()
+	w, err := databricks.NewWorkspaceClient()
+	require.NoError(t, err)
+
+	tmpDir := temporaryDbfsDir(t, w)
+
+	// create file hello
+	f, err := filer.NewDbfsClient(w, tmpDir)
+	require.NoError(t, err)
+	err = f.Write(ctx, "hello", strings.NewReader("abc"))
+	require.NoError(t, err)
+
+	// assert run fails
+	_, _, err = RequireErrorRun(t, "fs", "mkdir", "dbfs:"+path.Join(tmpDir, "hello"))
+	assert.ErrorContains(t, err, "Cannot create directory")
+}