mirror of https://github.com/databricks/cli.git
Compare commits
6 Commits
edce2aca6c
...
373508425f
Author | SHA1 | Date |
---|---|---|
shreyas-goenka | 373508425f | |
Pieter Noordhuis | abfd1713e0 | |
Pieter Noordhuis | a3cea07c9e | |
shreyas-goenka | abc2f3c825 | |
shreyas-goenka | c2e2abcc35 | |
Shreyas Goenka | be2d802d13 |
|
@ -12,6 +12,7 @@ import (
|
|||
"github.com/databricks/cli/libs/dbr"
|
||||
"github.com/databricks/cli/libs/diag"
|
||||
"github.com/databricks/cli/libs/dyn"
|
||||
"github.com/databricks/cli/libs/log"
|
||||
"github.com/databricks/cli/libs/textutil"
|
||||
"github.com/databricks/databricks-sdk-go/service/catalog"
|
||||
"github.com/databricks/databricks-sdk-go/service/jobs"
|
||||
|
@ -189,6 +190,14 @@ func (m *applyPresets) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnos
|
|||
diags = diags.Extend(diag.Errorf("schema %s is not defined", key))
|
||||
continue
|
||||
}
|
||||
|
||||
// If the catalog is already namespaced to the current user, we don't need
|
||||
// to prefix the schema name since it already falls under the user's namespace.
|
||||
if containsUserIdentity(s.CatalogName, b.Config.Workspace.CurrentUser) {
|
||||
log.Debugf(ctx, "Skipping schema %s since catalog %s already contains the user's identity", s.Name, s.CatalogName)
|
||||
continue
|
||||
}
|
||||
|
||||
s.Name = normalizePrefix(prefix) + s.Name
|
||||
// HTTP API for schemas doesn't yet support tags. It's only supported in
|
||||
// the Databricks UI and via the SQL API.
|
||||
|
|
|
@ -11,6 +11,7 @@ import (
|
|||
"github.com/databricks/cli/bundle/config/resources"
|
||||
"github.com/databricks/cli/libs/dbr"
|
||||
"github.com/databricks/databricks-sdk-go/service/catalog"
|
||||
"github.com/databricks/databricks-sdk-go/service/iam"
|
||||
"github.com/databricks/databricks-sdk-go/service/jobs"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
@ -98,12 +99,53 @@ func TestApplyPresetsPrefixForUcSchema(t *testing.T) {
|
|||
},
|
||||
want: "schema1",
|
||||
},
|
||||
{
|
||||
name: "skip prefix because catalog contains short name",
|
||||
prefix: "[prefix]",
|
||||
schema: &resources.Schema{
|
||||
CreateSchema: &catalog.CreateSchema{
|
||||
Name: "schema1",
|
||||
CatalogName: "dev_john_smith_test_catalog",
|
||||
},
|
||||
},
|
||||
want: "schema1",
|
||||
},
|
||||
{
|
||||
name: "skip prefix because catalog contains email",
|
||||
prefix: "[prefix]",
|
||||
schema: &resources.Schema{
|
||||
CreateSchema: &catalog.CreateSchema{
|
||||
Name: "schema1",
|
||||
CatalogName: "dev_john.smith@databricks.com_test_catalog",
|
||||
},
|
||||
},
|
||||
want: "schema1",
|
||||
},
|
||||
{
|
||||
name: "add prefix because catalog is not namespaced to user",
|
||||
prefix: "[prefix]",
|
||||
schema: &resources.Schema{
|
||||
CreateSchema: &catalog.CreateSchema{
|
||||
Name: "schema1",
|
||||
CatalogName: "test_catalog",
|
||||
},
|
||||
},
|
||||
want: "prefix_schema1",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
b := &bundle.Bundle{
|
||||
Config: config.Root{
|
||||
Workspace: config.Workspace{
|
||||
CurrentUser: &config.User{
|
||||
ShortName: "john_smith",
|
||||
User: &iam.User{
|
||||
UserName: "john.smith@databricks.com",
|
||||
},
|
||||
},
|
||||
},
|
||||
Resources: config.Resources{
|
||||
Schemas: map[string]*resources.Schema{
|
||||
"schema1": tt.schema,
|
||||
|
|
|
@ -72,6 +72,10 @@ func transformDevelopmentMode(ctx context.Context, b *bundle.Bundle) {
|
|||
}
|
||||
}
|
||||
|
||||
func containsUserIdentity(s string, u *config.User) bool {
|
||||
return strings.Contains(s, u.ShortName) || strings.Contains(s, u.UserName)
|
||||
}
|
||||
|
||||
func validateDevelopmentMode(b *bundle.Bundle) diag.Diagnostics {
|
||||
var diags diag.Diagnostics
|
||||
p := b.Config.Presets
|
||||
|
@ -101,7 +105,7 @@ func validateDevelopmentMode(b *bundle.Bundle) diag.Diagnostics {
|
|||
diags = diags.Extend(diag.Errorf("%s must start with '~/' or contain the current username to ensure uniqueness when using 'mode: development'", path))
|
||||
}
|
||||
}
|
||||
if p.NamePrefix != "" && !strings.Contains(p.NamePrefix, u.ShortName) && !strings.Contains(p.NamePrefix, u.UserName) {
|
||||
if p.NamePrefix != "" && !containsUserIdentity(p.NamePrefix, u) {
|
||||
// Resources such as pipelines require a unique name, e.g. '[dev steve] my_pipeline'.
|
||||
// For this reason we require the name prefix to contain the current username;
|
||||
// it's a pitfall for users if they don't include it and later find out that
|
||||
|
|
|
@ -126,7 +126,33 @@ func (t *translateContext) rewritePath(
|
|||
func (t *translateContext) translateNotebookPath(literal, localFullPath, localRelPath, remotePath string) (string, error) {
|
||||
nb, _, err := notebook.DetectWithFS(t.b.SyncRoot, filepath.ToSlash(localRelPath))
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return "", fmt.Errorf("notebook %s not found", literal)
|
||||
if filepath.Ext(localFullPath) != notebook.ExtensionNone {
|
||||
return "", fmt.Errorf("notebook %s not found", literal)
|
||||
}
|
||||
|
||||
extensions := []string{
|
||||
notebook.ExtensionPython,
|
||||
notebook.ExtensionR,
|
||||
notebook.ExtensionScala,
|
||||
notebook.ExtensionSql,
|
||||
notebook.ExtensionJupyter,
|
||||
}
|
||||
|
||||
// Check whether a file with a notebook extension already exists. This
|
||||
// way we can provide a more targeted error message.
|
||||
for _, ext := range extensions {
|
||||
literalWithExt := literal + ext
|
||||
localRelPathWithExt := filepath.ToSlash(localRelPath + ext)
|
||||
if _, err := fs.Stat(t.b.SyncRoot, localRelPathWithExt); err == nil {
|
||||
return "", fmt.Errorf(`notebook %s not found. Did you mean %s?
|
||||
Local notebook references are expected to contain one of the following
|
||||
file extensions: [%s]`, literal, literalWithExt, strings.Join(extensions, ", "))
|
||||
}
|
||||
}
|
||||
|
||||
// Return a generic error message if no matching possible file is found.
|
||||
return "", fmt.Errorf(`notebook %s not found. Local notebook references are expected
|
||||
to contain one of the following file extensions: [%s]`, literal, strings.Join(extensions, ", "))
|
||||
}
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("unable to determine if %s is a notebook: %w", localFullPath, err)
|
||||
|
|
|
@ -2,6 +2,7 @@ package mutator_test
|
|||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
|
@ -508,6 +509,59 @@ func TestPipelineNotebookDoesNotExistError(t *testing.T) {
|
|||
assert.EqualError(t, diags.Error(), "notebook ./doesnt_exist.py not found")
|
||||
}
|
||||
|
||||
func TestPipelineNotebookDoesNotExistErrorWithoutExtension(t *testing.T) {
|
||||
for _, ext := range []string{
|
||||
".py",
|
||||
".r",
|
||||
".scala",
|
||||
".sql",
|
||||
".ipynb",
|
||||
"",
|
||||
} {
|
||||
t.Run("case_"+ext, func(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
|
||||
if ext != "" {
|
||||
touchEmptyFile(t, filepath.Join(dir, "foo"+ext))
|
||||
}
|
||||
|
||||
b := &bundle.Bundle{
|
||||
SyncRootPath: dir,
|
||||
SyncRoot: vfs.MustNew(dir),
|
||||
Config: config.Root{
|
||||
Resources: config.Resources{
|
||||
Pipelines: map[string]*resources.Pipeline{
|
||||
"pipeline": {
|
||||
PipelineSpec: &pipelines.PipelineSpec{
|
||||
Libraries: []pipelines.PipelineLibrary{
|
||||
{
|
||||
Notebook: &pipelines.NotebookLibrary{
|
||||
Path: "./foo",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
bundletest.SetLocation(b, ".", []dyn.Location{{File: filepath.Join(dir, "fake.yml")}})
|
||||
diags := bundle.Apply(context.Background(), b, mutator.TranslatePaths())
|
||||
|
||||
if ext == "" {
|
||||
assert.EqualError(t, diags.Error(), `notebook ./foo not found. Local notebook references are expected
|
||||
to contain one of the following file extensions: [.py, .r, .scala, .sql, .ipynb]`)
|
||||
} else {
|
||||
assert.EqualError(t, diags.Error(), fmt.Sprintf(`notebook ./foo not found. Did you mean ./foo%s?
|
||||
Local notebook references are expected to contain one of the following
|
||||
file extensions: [.py, .r, .scala, .sql, .ipynb]`, ext))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPipelineFileDoesNotExistError(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
|
||||
|
|
|
@ -21,6 +21,12 @@ func (v *filesToSync) Name() string {
|
|||
}
|
||||
|
||||
func (v *filesToSync) Apply(ctx context.Context, rb bundle.ReadOnlyBundle) diag.Diagnostics {
|
||||
// The user may be intentional about not synchronizing any files.
|
||||
// In this case, we should not show any warnings.
|
||||
if len(rb.Config().Sync.Paths) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
sync, err := files.GetSync(ctx, rb)
|
||||
if err != nil {
|
||||
return diag.FromErr(err)
|
||||
|
@ -31,6 +37,7 @@ func (v *filesToSync) Apply(ctx context.Context, rb bundle.ReadOnlyBundle) diag.
|
|||
return diag.FromErr(err)
|
||||
}
|
||||
|
||||
// If there are files to sync, we don't need to show any warnings.
|
||||
if len(fl) != 0 {
|
||||
return nil
|
||||
}
|
||||
|
|
|
@ -0,0 +1,105 @@
|
|||
package validate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/databricks/cli/bundle"
|
||||
"github.com/databricks/cli/bundle/config"
|
||||
"github.com/databricks/cli/internal/testutil"
|
||||
"github.com/databricks/cli/libs/diag"
|
||||
"github.com/databricks/cli/libs/vfs"
|
||||
sdkconfig "github.com/databricks/databricks-sdk-go/config"
|
||||
"github.com/databricks/databricks-sdk-go/experimental/mocks"
|
||||
"github.com/databricks/databricks-sdk-go/service/iam"
|
||||
"github.com/databricks/databricks-sdk-go/service/workspace"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestFilesToSync_NoPaths(t *testing.T) {
|
||||
b := &bundle.Bundle{
|
||||
Config: config.Root{
|
||||
Sync: config.Sync{
|
||||
Paths: []string{},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
rb := bundle.ReadOnly(b)
|
||||
diags := bundle.ApplyReadOnly(ctx, rb, FilesToSync())
|
||||
assert.Empty(t, diags)
|
||||
}
|
||||
|
||||
func setupBundleForFilesToSyncTest(t *testing.T) *bundle.Bundle {
|
||||
dir := t.TempDir()
|
||||
|
||||
testutil.Touch(t, dir, "file1")
|
||||
testutil.Touch(t, dir, "file2")
|
||||
|
||||
b := &bundle.Bundle{
|
||||
BundleRootPath: dir,
|
||||
BundleRoot: vfs.MustNew(dir),
|
||||
SyncRootPath: dir,
|
||||
SyncRoot: vfs.MustNew(dir),
|
||||
Config: config.Root{
|
||||
Bundle: config.Bundle{
|
||||
Target: "default",
|
||||
},
|
||||
Workspace: config.Workspace{
|
||||
FilePath: "/this/doesnt/matter",
|
||||
CurrentUser: &config.User{
|
||||
User: &iam.User{},
|
||||
},
|
||||
},
|
||||
Sync: config.Sync{
|
||||
// Paths are relative to [SyncRootPath].
|
||||
Paths: []string{"."},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
m := mocks.NewMockWorkspaceClient(t)
|
||||
m.WorkspaceClient.Config = &sdkconfig.Config{
|
||||
Host: "https://foo.com",
|
||||
}
|
||||
|
||||
// The initialization logic in [sync.New] performs a check on the destination path.
|
||||
// Removing this check at initialization time is tbd...
|
||||
m.GetMockWorkspaceAPI().EXPECT().GetStatusByPath(mock.Anything, "/this/doesnt/matter").Return(&workspace.ObjectInfo{
|
||||
ObjectType: workspace.ObjectTypeDirectory,
|
||||
}, nil)
|
||||
|
||||
b.SetWorkpaceClient(m.WorkspaceClient)
|
||||
return b
|
||||
}
|
||||
|
||||
func TestFilesToSync_EverythingIgnored(t *testing.T) {
|
||||
b := setupBundleForFilesToSyncTest(t)
|
||||
|
||||
// Ignore all files.
|
||||
testutil.WriteFile(t, "*\n.*\n", b.BundleRootPath, ".gitignore")
|
||||
|
||||
ctx := context.Background()
|
||||
rb := bundle.ReadOnly(b)
|
||||
diags := bundle.ApplyReadOnly(ctx, rb, FilesToSync())
|
||||
require.Equal(t, 1, len(diags))
|
||||
assert.Equal(t, diag.Warning, diags[0].Severity)
|
||||
assert.Equal(t, "There are no files to sync, please check your .gitignore", diags[0].Summary)
|
||||
}
|
||||
|
||||
func TestFilesToSync_EverythingExcluded(t *testing.T) {
|
||||
b := setupBundleForFilesToSyncTest(t)
|
||||
|
||||
// Exclude all files.
|
||||
b.Config.Sync.Exclude = []string{"*"}
|
||||
|
||||
ctx := context.Background()
|
||||
rb := bundle.ReadOnly(b)
|
||||
diags := bundle.ApplyReadOnly(ctx, rb, FilesToSync())
|
||||
require.Equal(t, 1, len(diags))
|
||||
assert.Equal(t, diag.Warning, diags[0].Severity)
|
||||
assert.Equal(t, "There are no files to sync, please check your .gitignore and sync.exclude configuration", diags[0].Summary)
|
||||
}
|
|
@ -22,6 +22,8 @@ type Lookup struct {
|
|||
|
||||
Metastore string `json:"metastore,omitempty"`
|
||||
|
||||
NotificationDestination string `json:"notification_destination,omitempty"`
|
||||
|
||||
Pipeline string `json:"pipeline,omitempty"`
|
||||
|
||||
Query string `json:"query,omitempty"`
|
||||
|
@ -63,6 +65,9 @@ func (l *Lookup) constructResolver() (resolver, error) {
|
|||
if l.Metastore != "" {
|
||||
resolvers = append(resolvers, resolveMetastore{name: l.Metastore})
|
||||
}
|
||||
if l.NotificationDestination != "" {
|
||||
resolvers = append(resolvers, resolveNotificationDestination{name: l.NotificationDestination})
|
||||
}
|
||||
if l.Pipeline != "" {
|
||||
resolvers = append(resolvers, resolvePipeline{name: l.Pipeline})
|
||||
}
|
||||
|
|
|
@ -0,0 +1,46 @@
|
|||
package variable
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/databricks/databricks-sdk-go"
|
||||
"github.com/databricks/databricks-sdk-go/service/settings"
|
||||
)
|
||||
|
||||
type resolveNotificationDestination struct {
|
||||
name string
|
||||
}
|
||||
|
||||
func (l resolveNotificationDestination) Resolve(ctx context.Context, w *databricks.WorkspaceClient) (string, error) {
|
||||
result, err := w.NotificationDestinations.ListAll(ctx, settings.ListNotificationDestinationsRequest{
|
||||
// The default page size for this API is 20.
|
||||
// We use a higher value to make fewer API calls.
|
||||
PageSize: 200,
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
// Collect all notification destinations with the given name.
|
||||
var entities []settings.ListNotificationDestinationsResult
|
||||
for _, entity := range result {
|
||||
if entity.DisplayName == l.name {
|
||||
entities = append(entities, entity)
|
||||
}
|
||||
}
|
||||
|
||||
// Return the ID of the first matching notification destination.
|
||||
switch len(entities) {
|
||||
case 0:
|
||||
return "", fmt.Errorf("notification destination named %q does not exist", l.name)
|
||||
case 1:
|
||||
return entities[0].Id, nil
|
||||
default:
|
||||
return "", fmt.Errorf("there are %d instances of clusters named %q", len(entities), l.name)
|
||||
}
|
||||
}
|
||||
|
||||
func (l resolveNotificationDestination) String() string {
|
||||
return fmt.Sprintf("notification-destination: %s", l.name)
|
||||
}
|
|
@ -0,0 +1,82 @@
|
|||
package variable
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/databricks/databricks-sdk-go/experimental/mocks"
|
||||
"github.com/databricks/databricks-sdk-go/service/settings"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestResolveNotificationDestination_ResolveSuccess(t *testing.T) {
|
||||
m := mocks.NewMockWorkspaceClient(t)
|
||||
|
||||
api := m.GetMockNotificationDestinationsAPI()
|
||||
api.EXPECT().
|
||||
ListAll(mock.Anything, mock.Anything).
|
||||
Return([]settings.ListNotificationDestinationsResult{
|
||||
{Id: "1234", DisplayName: "destination"},
|
||||
}, nil)
|
||||
|
||||
ctx := context.Background()
|
||||
l := resolveNotificationDestination{name: "destination"}
|
||||
result, err := l.Resolve(ctx, m.WorkspaceClient)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "1234", result)
|
||||
}
|
||||
|
||||
func TestResolveNotificationDestination_ResolveError(t *testing.T) {
|
||||
m := mocks.NewMockWorkspaceClient(t)
|
||||
|
||||
api := m.GetMockNotificationDestinationsAPI()
|
||||
api.EXPECT().
|
||||
ListAll(mock.Anything, mock.Anything).
|
||||
Return(nil, fmt.Errorf("bad"))
|
||||
|
||||
ctx := context.Background()
|
||||
l := resolveNotificationDestination{name: "destination"}
|
||||
_, err := l.Resolve(ctx, m.WorkspaceClient)
|
||||
assert.ErrorContains(t, err, "bad")
|
||||
}
|
||||
|
||||
func TestResolveNotificationDestination_ResolveNotFound(t *testing.T) {
|
||||
m := mocks.NewMockWorkspaceClient(t)
|
||||
|
||||
api := m.GetMockNotificationDestinationsAPI()
|
||||
api.EXPECT().
|
||||
ListAll(mock.Anything, mock.Anything).
|
||||
Return([]settings.ListNotificationDestinationsResult{}, nil)
|
||||
|
||||
ctx := context.Background()
|
||||
l := resolveNotificationDestination{name: "destination"}
|
||||
_, err := l.Resolve(ctx, m.WorkspaceClient)
|
||||
require.Error(t, err)
|
||||
assert.ErrorContains(t, err, `notification destination named "destination" does not exist`)
|
||||
}
|
||||
|
||||
func TestResolveNotificationDestination_ResolveMultiple(t *testing.T) {
|
||||
m := mocks.NewMockWorkspaceClient(t)
|
||||
|
||||
api := m.GetMockNotificationDestinationsAPI()
|
||||
api.EXPECT().
|
||||
ListAll(mock.Anything, mock.Anything).
|
||||
Return([]settings.ListNotificationDestinationsResult{
|
||||
{Id: "1234", DisplayName: "destination"},
|
||||
{Id: "5678", DisplayName: "destination"},
|
||||
}, nil)
|
||||
|
||||
ctx := context.Background()
|
||||
l := resolveNotificationDestination{name: "destination"}
|
||||
_, err := l.Resolve(ctx, m.WorkspaceClient)
|
||||
require.Error(t, err)
|
||||
assert.ErrorContains(t, err, `there are 2 instances of clusters named "destination"`)
|
||||
}
|
||||
|
||||
func TestResolveNotificationDestination_String(t *testing.T) {
|
||||
l := resolveNotificationDestination{name: "name"}
|
||||
assert.Equal(t, "notification-destination: name", l.String())
|
||||
}
|
|
@ -97,7 +97,7 @@ func TestAccBundleInitOnMlopsStacks(t *testing.T) {
|
|||
require.NoError(t, err)
|
||||
job, err := w.Jobs.GetByJobId(context.Background(), batchJobId)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, fmt.Sprintf("dev-%s-batch-inference-job", projectName), job.Settings.Name)
|
||||
assert.Contains(t, job.Settings.Name, fmt.Sprintf("dev-%s-batch-inference-job", projectName))
|
||||
}
|
||||
|
||||
func TestAccBundleInitHelpers(t *testing.T) {
|
||||
|
|
Loading…
Reference in New Issue