acc: nostamp in Python, and drop it where the stamp cannot appear

acceptance/bin/nostamp

@@ -1,41 +1,71 @@
-#!/usr/bin/env bash
-
-# Reads JSON on stdin, writes it back with the deployment stamp removed.
-#
-# Deployment history recording adds deployment_id and version_id to every job and
-# pipeline. Acceptance tests compare output byte for byte, so those two extra fields
-# would fail every test in the DATABRICKS_BUNDLE_RECORD_DEPLOYMENT_HISTORY=true run
-# (see bundle/test.toml).
-# Pipe a plan, a state dump, or a resource payload through this and the test asserts one
-# golden file either way. Tests under bundle/dms assert the stamp itself and must not.
-#
-# Three passes, because the stamp shows up in three shapes:
-#
-#   1. Nested in a deployment block, as printed by `jobs get` / `pipelines get`:
-#        "deployment": {"deployment_id": "87..", "kind": "BUNDLE",
-#                       "metadata_file_path": "/x", "version_id": "1"}
-#      -> both keys dropped, "kind" and "metadata_file_path" kept.
-#
-#   2. Flat in a plan's "changes", keyed by field path:
-#        "changes": {"deployment.version_id": {"action": "skip", ...}, "name": {...}}
-#      -> the stamp entries dropped, real changes kept.
-#
-#   3. A "changes" object that pass 2 emptied to {} - it existed only because of
-#      recording, so the key goes too.
-#
-# Two things it deliberately keeps:
-#
-#   - A deployment_id anywhere else. Pass 1 requires "kind" and "metadata_file_path" as
-#     neighbours, a pair unique to the deployment block, so an unrelated field of the
-#     same name is untouched.
-#   - "version_id": "". A terraform state dump carries that for a job it never stamped,
-#     and it is the test's own expected output, so `.value == ""` keeps it.
-#
-# Arguments are passed to jq, for callers whose input is not formatted the way jq
-# formats by default: a state dump is printed verbatim from disk, so it needs
-# --indent 1 to come back out unchanged.
-jq "$@" '((.. | objects | select(has("kind") and has("metadata_file_path")))
-     |= with_entries(select((.key | IN("deployment_id", "version_id")) == false or .value == "")))
-    | ((.. | objects | .changes? | objects)
-       |= with_entries(select(.key | IN("deployment.deployment_id", "deployment.version_id") | not)))
-    | del(.. | objects | select(.changes == {}) | .changes)'
+#!/usr/bin/env python3
+"""Read JSON on stdin, write it back with the deployment stamp removed.
+
+Deployment history recording adds deployment_id and version_id to every job and pipeline.
+Acceptance tests compare output byte for byte, so those two fields would fail every test in
+the DMS=true run (see bundle/test.toml). Pipe a plan, a state dump or a resource payload
+through this and the test asserts one golden file either way. Tests under bundle/dms assert
+the stamp itself and must not use this.
+"""
+
+import json
+import sys
+
+from print_requests import read_json_many
+
+STAMP_KEYS = ("deployment_id", "version_id")
+
+# The same two fields keyed by path, as a plan's "changes" map writes them.
+STAMP_PATHS = ("deployment.deployment_id", "deployment.version_id")
+
+
+def is_deployment_block(obj):
+    """A deployment block, as `jobs get` and `pipelines get` print it. Both neighbours are
+    required so an unrelated field named deployment_id is left alone."""
+    return "kind" in obj and "metadata_file_path" in obj
+
+
+def strip(node):
+    """Remove the stamp from every shape it appears in, in place."""
+    if isinstance(node, list):
+        for item in node:
+            strip(item)
+        return
+
+    if not isinstance(node, dict):
+        return
+
+    for value in node.values():
+        strip(value)
+
+    if is_deployment_block(node):
+        for key in STAMP_KEYS:
+            # A terraform state dump carries version_id "" for a job it never stamped, and
+            # that is the test's own expected output, so only a real id is dropped.
+            if node.get(key) != "":
+                node.pop(key, None)
+
+    changes = node.get("changes")
+    if isinstance(changes, dict):
+        for path in STAMP_PATHS:
+            changes.pop(path, None)
+        # The map existed only because of recording, so the key goes with its last entry.
+        if not changes:
+            del node["changes"]
+
+
+def main():
+    if len(sys.argv) > 1:
+        sys.exit(f"nostamp reads JSON on stdin and takes no arguments, got: {' '.join(sys.argv[1:])}")
+
+    data = sys.stdin.read()
+    if not data.strip():
+        return
+
+    for doc in read_json_many(data):
+        strip(doc)
+        print(json.dumps(doc, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

acceptance/bundle/state/permission_level_migration/script

@@ -15,4 +15,4 @@ title "Deploy (migrates state)"
 trace $CLI bundle deploy
 
 title "Print state after deploy"
-trace print_state.py | nostamp --indent 1
+trace print_state.py