Add acceptance test for bundle run

.wsignore

@@ -19,6 +19,7 @@ acceptance/selftest/record_cloud/volume-io/hello.txt
 
 # "bundle run" has trailing whitespace:
 acceptance/bundle/integration_whl/*/output.txt
+acceptance/bundle/run/basic/output.txt
 
 # "bundle init" has trailing whitespace:
 acceptance/bundle/templates-machinery/helpers-error/output.txt

acceptance/bundle/run/basic/databricks.yml

@@ -0,0 +1,18 @@
+bundle:
+  name: caterpillar
+
+
+resources:
+  jobs:
+    foo:
+      name: foo
+      tasks:
+        - task_key: task
+          spark_python_task:
+            python_file: ./foo.py
+          environment_key: default
+
+      environments:
+        - environment_key: default
+          spec:
+            client: "2"

acceptance/bundle/run/basic/foo.py

@@ -0,0 +1 @@
+print(1)

acceptance/bundle/run/basic/output.txt

@@ -0,0 +1,34 @@
+
+=== no run key specified
+>>> [CLI] bundle run
+Error: expected a KEY of the resource to run
+
+Exit code: 1
+
+=== deploy and run resource
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/caterpillar/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle run foo
+Run URL: [DATABRICKS_URL]/job/run/1
+
+[DATE] HH:MM:SS "run-name" TERMINATED 
+
+=== no resource key with --
+>>> [CLI] bundle run --
+Error: expected a KEY of the resource to run
+
+Exit code: 1
+
+=== resource key with parameters
+>>> [CLI] bundle run foo -- arg1 arg2
+Run URL: [DATABRICKS_URL]/job/run/2
+
+[DATE] HH:MM:SS "run-name" TERMINATED 
+
+=== inline script
+>>> [CLI] bundle run -- echo hello
+hello

acceptance/bundle/run/basic/script

@@ -0,0 +1,16 @@
+title "no run key specified"
+errcode trace $CLI bundle run
+
+title "deploy and run resource"
+errcode trace $CLI bundle deploy
+
+errcode trace $CLI bundle run foo
+
+title "no resource key with --"
+errcode trace $CLI bundle run --
+
+title "resource key with parameters"
+errcode trace $CLI bundle run foo -- arg1 arg2
+
+title "inline script"
+errcode trace $CLI bundle run -- echo "hello"

acceptance/bundle/run/basic/test.toml

@@ -0,0 +1,7 @@
+[[Repls]]
+Old = "(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]"
+New = "HH:MM:SS"
+
+[[Repls]]
+Old = '20\d\d-\d\d-\d\d'
+New = '[DATE]'

acceptance/internal/handlers.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strconv"
 
 	"github.com/databricks/databricks-sdk-go/service/catalog"
 	"github.com/databricks/databricks-sdk-go/service/iam"
@@ -194,6 +195,31 @@ func addDefaultHandlers(server *testserver.Server) {
 		return req.Workspace.JobsList()
 	})
 
+	server.Handle("POST", "/api/2.2/jobs/run-now", func(req testserver.Request) any {
+		var request jobs.RunNow
+		if err := json.Unmarshal(req.Body, &request); err != nil {
+			return testserver.Response{
+				Body:       fmt.Sprintf("internal error: %s", err),
+				StatusCode: 500,
+			}
+		}
+
+		return req.Workspace.JobsRunNow(request.JobId)
+	})
+
+	server.Handle("GET", "/api/2.2/jobs/runs/get", func(req testserver.Request) any {
+		runId := req.URL.Query().Get("run_id")
+		runIdInt, err := strconv.ParseInt(runId, 10, 64)
+		if err != nil {
+			return testserver.Response{
+				Body:       fmt.Sprintf("internal error: %s", err),
+				StatusCode: 500,
+			}
+		}
+
+		return req.Workspace.JobsGetRun(runIdInt)
+	})
+
 	server.Handle("GET", "/oidc/.well-known/oauth-authorization-server", func(_ testserver.Request) any {
 		return map[string]string{
 			"authorization_endpoint": server.URL + "oidc/v1/authorize",

libs/testserver/fake_workspace.go

@@ -20,13 +20,16 @@ import (
 
 // FakeWorkspace holds a state of a workspace for acceptance tests.
 type FakeWorkspace struct {
-	mu sync.Mutex
+	mu  sync.Mutex
+	url string
 
 	directories map[string]bool
 	files       map[string][]byte
 	// normally, ids are not sequential, but we make them sequential for deterministic diff
-	nextJobId int64
-	jobs      map[int64]jobs.Job
+	nextJobId    int64
+	nextJobRunId int64
+	jobs         map[int64]jobs.Job
+	jobRuns      map[int64]jobs.Run
 
 	Pipelines map[string]pipelines.PipelineSpec
 	Monitors  map[string]catalog.MonitorInfo
@@ -70,19 +73,21 @@ func MapDelete[T any](w *FakeWorkspace, collection map[string]T, key string) Res
 	return Response{}
 }
 
-func NewFakeWorkspace() *FakeWorkspace {
+func NewFakeWorkspace(url string) *FakeWorkspace {
 	return &FakeWorkspace{
+		url: url,
 		directories: map[string]bool{
 			"/Workspace": true,
 		},
-		files:     map[string][]byte{},
-		jobs:      map[int64]jobs.Job{},
-		nextJobId: 1,
-
-		Pipelines: map[string]pipelines.PipelineSpec{},
-		Monitors:  map[string]catalog.MonitorInfo{},
-		Apps:      map[string]apps.App{},
-		Schemas:   map[string]catalog.SchemaInfo{},
+		files:        map[string][]byte{},
+		jobs:         map[int64]jobs.Job{},
+		jobRuns:      map[int64]jobs.Run{},
+		nextJobId:    1,
+		nextJobRunId: 1,
+		Pipelines:    map[string]pipelines.PipelineSpec{},
+		Monitors:     map[string]catalog.MonitorInfo{},
+		Apps:         map[string]apps.App{},
+		Schemas:      map[string]catalog.SchemaInfo{},
 	}
 }
 
@@ -247,6 +252,52 @@ func (s *FakeWorkspace) JobsGet(jobId string) Response {
 	}
 }
 
+func (s *FakeWorkspace) JobsRunNow(jobId int64) Response {
+	defer s.LockUnlock()()
+
+	_, ok := s.jobs[jobId]
+	if !ok {
+		return Response{
+			StatusCode: 404,
+		}
+	}
+
+	runId := s.nextJobRunId
+	s.nextJobRunId++
+	s.jobRuns[runId] = jobs.Run{
+		RunId: runId,
+		State: &jobs.RunState{
+			LifeCycleState: jobs.RunLifeCycleStateRunning,
+		},
+		RunPageUrl: fmt.Sprintf("%s/job/run/%d", s.url, runId),
+		RunType:    jobs.RunTypeJobRun,
+		RunName:    "run-name",
+	}
+
+	return Response{
+		Body: jobs.RunNowResponse{
+			RunId: runId,
+		},
+	}
+}
+
+func (s *FakeWorkspace) JobsGetRun(runId int64) Response {
+	defer s.LockUnlock()()
+
+	run, ok := s.jobRuns[runId]
+	if !ok {
+		return Response{
+			StatusCode: 404,
+		}
+	}
+
+	// Mark the run as terminated.
+	run.State.LifeCycleState = jobs.RunLifeCycleStateTerminated
+	return Response{
+		Body: run,
+	}
+}
+
 func (s *FakeWorkspace) PipelinesGet(pipelineId string) Response {
 	defer s.LockUnlock()()
 

libs/testserver/server.go

@@ -240,7 +240,7 @@ func (s *Server) getWorkspaceForToken(token string) *FakeWorkspace {
 	defer s.mu.Unlock()
 
 	if _, ok := s.fakeWorkspaces[token]; !ok {
-		s.fakeWorkspaces[token] = NewFakeWorkspace()
+		s.fakeWorkspaces[token] = NewFakeWorkspace(s.Server.URL)
 	}
 
 	return s.fakeWorkspaces[token]