acc: Add a test for immutable folders + DMS

acceptance/bundle/dms/immutable-folder-lifecycle/databricks.yml.tmpl

@@ -0,0 +1,20 @@
+bundle:
+  name: dms-immutable-folder-lifecycle-$UNIQUE_NAME
+
+experimental:
+  immutable_folder: true
+  deployment_history: true
+
+resources:
+  jobs:
+    foo:
+      name: foo
+      tasks:
+        - task_key: main
+          spark_python_task:
+            python_file: ./src/main.py
+          environment_key: env
+      environments:
+        - environment_key: env
+          spec:
+            environment_version: "4"

acceptance/bundle/dms/immutable-folder-lifecycle/out.test.toml

@@ -0,0 +1,2 @@
+Cloud = false
+EnvMatrix.DMS = ["true"]

acceptance/bundle/dms/immutable-folder-lifecycle/output.txt

@@ -0,0 +1,73 @@
+
+=== Deploy stages an operation for the job and for the immutable folder snapshot
+>>> [CLI] bundle deploy
+Created internal_immutable_snapshots.immutable
+Created jobs.foo
+Files: 0 uploaded, 0 deleted
+Resources: 2 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py --dms --method POST //versions ^//complete ^//heartbeat
+{
+  "method": "POST",
+  "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions",
+  "q": {
+    "version_id": "1"
+  },
+  "body": {
+    "cli_version": "[CLI_VERSION]",
+    "operations": [
+      {
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "internal_immutable_snapshots.immutable"
+      },
+      {
+        "action_type": "OPERATION_ACTION_TYPE_CREATE",
+        "resource_key": "jobs.foo"
+      }
+    ],
+    "version_type": "VERSION_TYPE_DEPLOY"
+  }
+}
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
+[
+  {
+    "last_action_type": "OPERATION_ACTION_TYPE_CREATE",
+    "last_version_id": "1",
+    "resource_key": "internal_immutable_snapshots.immutable"
+  },
+  {
+    "last_action_type": "OPERATION_ACTION_TYPE_CREATE",
+    "last_version_id": "1",
+    "resource_key": "jobs.foo"
+  }
+]
+
+=== A changed file recreates the snapshot, and both resources are recorded at version 2
+>>> [CLI] bundle deploy
+Recreated internal_immutable_snapshots.immutable
+Updated jobs.foo
+Files: 0 uploaded, 0 deleted
+Resources: 1 created, 1 changed, 1 deleted, 0 unchanged
+
+>>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
+[
+  {
+    "last_action_type": "OPERATION_ACTION_TYPE_RECREATE",
+    "last_version_id": "2",
+    "resource_key": "internal_immutable_snapshots.immutable"
+  },
+  {
+    "last_action_type": "OPERATION_ACTION_TYPE_UPDATE",
+    "last_version_id": "2",
+    "resource_key": "jobs.foo"
+  }
+]
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.jobs.foo
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/dms-immutable-folder-lifecycle-[UNIQUE_NAME]/default
+
+Destroy: 1 deleted

acceptance/bundle/dms/immutable-folder-lifecycle/script

@@ -0,0 +1,23 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    rm -f "$OUT_REQUESTS"
+}
+trap cleanup EXIT
+
+list_resources() {
+    deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-immutable-folder-lifecycle-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
+    add_repl "$deployment_id" DEPLOYMENT_ID
+    trace MSYS_NO_PATHCONV=1 $CLI bundle-deployments list-resources "deployments/${deployment_id}" -o json | jq -S 'map({resource_key, last_version_id, last_action_type})'
+}
+
+title "Deploy stages an operation for the job and for the immutable folder snapshot"
+trace $CLI bundle deploy
+trace print_requests.py --dms --method POST //versions '^//complete' '^//heartbeat'
+list_resources
+
+title "A changed file recreates the snapshot, and both resources are recorded at version 2"
+echo 'print("changed")' > src/main.py
+trace $CLI bundle deploy
+list_resources

acceptance/bundle/dms/immutable-folder-lifecycle/src/main.py

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

acceptance/bundle/dms/immutable-folder-lifecycle/test.toml

@@ -0,0 +1,6 @@
+Cloud = false
+
+# The snapshot path is content-addressed.
+[[Repls]]
+Old = '[0-9a-f]{64}'
+New = '[SNAPSHOT_HASH]'

libs/testserver/bundledeployments.go

@@ -34,6 +34,53 @@ const dmsDeploymentNodeName = "resources.deployment.json"
 // created.
 var dmsUpdatableOperationFields = []string{"state", "error_message", "resource_id", "status"}
 
+// dmsResourceTypes are the resource_key prefixes the service maps to a resource type. A staged
+// operation with any other prefix is rejected, so a resource type the service does not know
+// cannot be recorded. Mirrors PREFIX_MAP in universe's
+// bundle/metadata-service/src/main/java/com/databricks/deploymentmetadataservice/parser/ResourceTypes.java.
+var dmsResourceTypes = map[string]bool{
+	"jobs":                         true,
+	"pipelines":                    true,
+	"models":                       true,
+	"registered_models":            true,
+	"experiments":                  true,
+	"model_serving_endpoints":      true,
+	"quality_monitors":             true,
+	"schemas":                      true,
+	"volumes":                      true,
+	"clusters":                     true,
+	"dashboards":                   true,
+	"apps":                         true,
+	"catalogs":                     true,
+	"external_locations":           true,
+	"secret_scopes":                true,
+	"alerts":                       true,
+	"sql_warehouses":               true,
+	"database_instances":           true,
+	"database_catalogs":            true,
+	"synced_database_tables":       true,
+	"postgres_projects":            true,
+	"postgres_branches":            true,
+	"postgres_endpoints":           true,
+	"postgres_catalogs":            true,
+	"postgres_databases":           true,
+	"postgres_roles":               true,
+	"postgres_synced_tables":       true,
+	"vector_search_endpoints":      true,
+	"vector_search_indexes":        true,
+	"job_runs":                     true,
+	"genie_spaces":                 true,
+	"instance_pools":               true,
+	"secrets":                      true,
+	"cluster_policies":             true,
+	"postgres_snapshot_schedules":  true,
+	"model_services":               true,
+	"mcp_services":                 true,
+	"model_provider_services":      true,
+	"agent_services":               true,
+	"internal_immutable_snapshots": true,
+}
+
 // DmsDeployment holds a deployment record together with the versions and
 // resources recorded under it, so the read APIs (ListVersions/ListResources)
 // can serve back what deploys wrote.
@@ -330,7 +377,7 @@ func (s *FakeWorkspace) CreateVersion(req Request, deploymentID string) Response
 		switch {
 		case staged.ResourceKey == "":
 			return dmsInvalidArgument("operations.resource_key is required")
-		case !strings.Contains(staged.ResourceKey, "."):
+		case !hasKnownResourceType(staged.ResourceKey):
 			return dmsInvalidArgument("operations.resource_key must have a known resource type prefix (e.g. 'jobs.', 'pipelines.'): " + staged.ResourceKey)
 		case staged.ActionType == "":
 			return dmsInvalidArgument("operations.action_type is required and must not be UNSPECIFIED for resource " + staged.ResourceKey)
@@ -635,6 +682,13 @@ func (s *FakeWorkspace) ListOperations(deploymentID, versionID string) Response
 	return Response{Body: response}
 }
 
+// hasKnownResourceType reports whether the part of resourceKey before its first '.' is a resource
+// type the service knows.
+func hasKnownResourceType(resourceKey string) bool {
+	prefix, _, found := strings.Cut(resourceKey, ".")
+	return found && dmsResourceTypes[prefix]
+}
+
 // leavesLiveResource reports whether a failed operation of this type leaves a resource behind
 // that it still has to describe. A create leaves nothing, and a recreate already deleted.
 func leavesLiveResource(action bundledeployments.OperationActionType) bool {