Add test recording permissions API behaviour (#3900)

## Changes
Add test recording permissions API behaviour wrt multiple levels for the
same principal.

## Why
Document behaviour (important for permissions processing). I also plan
to use this test to fix the testserver.

---------

Co-authored-by: shreyas-goenka <[email protected]>

Author: denik

acceptance/bundle/resources/permissions/factcheck/check_permissions.py

@@ -0,0 +1,80 @@
+import sys
+import os
+import pprint
+import json
+import subprocess
+
+
+cli = os.environ["CLI"]
+
+
+class Error(Exception):
+    pass
+
+
+def run_json(cmd):
+    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8")
+    if result.returncode != 0:
+        if result.stderr.strip():
+            raise Error(result.stderr.strip())
+        raise Error(f"{cmd} failed with code {result.returncode}")
+    try:
+        return json.loads(result.stdout)
+    except Exception as ex:
+        raise Error(f"{cmd} returned non-json: {ex}\n{result.stdout}")
+
+
+def test_permissions(target_user, resource_type, resource_id, levels, expected):
+    """Set permissions on a given resource/principal with a given set of levels. Verify that resulting levels match expected."""
+    acls = []
+    if resource_type == "jobs" and target_user != os.environ["CURRENT_USER_NAME"]:
+        # make sure we have IS_OWNER, otherwise backend returns an error
+        acls.append({"service_principal_name": os.environ["CURRENT_USER_NAME"], "permission_level": "IS_OWNER"})
+
+    for level in levels.split(","):
+        acls.append({"user_name": target_user, "permission_level": level})
+
+    request = {"access_control_list": acls}
+
+    try:
+        set_result = run_json([cli, "permissions", "set", resource_type, resource_id, "--json", json.dumps(request)])
+    except Error as ex:
+        sys.exit(f"{resource_type} {levels} => SET ERROR {ex}\nREQUEST:\n" + pprint.pformat(request))
+
+    get_result = run_json([cli, "permissions", "get", resource_type, resource_id])
+
+    if set_result != get_result:
+        print("set() response is different from get() response")
+        print("set:")
+        pprint.pprint(set_result)
+        print("get:")
+        pprint.pprint(get_result)
+
+    resulting_levels = []
+
+    for item in set_result["access_control_list"]:
+        if (item.get("user_name") or item.get("service_principal_name")) != target_user:
+            continue
+        for perm in item["all_permissions"]:
+            if perm.get("inherited"):
+                continue
+            resulting_levels.append(perm["permission_level"])
+
+    resulting_levels = ", ".join(resulting_levels)
+    if expected == resulting_levels:
+        print(f"{resource_type} {levels} => {resulting_levels}")
+    else:
+        print(f"{resource_type} {levels} => {resulting_levels}; EXPECTED: {expected}")
+        print("REQUEST")
+        pprint.pprint(request)
+        print("RESPONSE")
+        pprint.pprint(set_result)
+        print()
+
+
+def main():
+    test_permissions(*sys.argv[1:])
+
+
+if __name__ == "__main__":
+    main()

acceptance/bundle/resources/permissions/factcheck/databricks.yml.tmpl

@@ -0,0 +1,22 @@
+bundle:
+  name: test_bundle_$UNIQUE_NAME
+
+resources:
+  jobs:
+    job:
+      name: job1
+      tasks:
+        - task_key: main
+          notebook_task:
+            notebook_path: /Workspace/Users/[email protected]/notebook
+            source: WORKSPACE
+          new_cluster:
+            num_workers: 1
+            num_workers: 1
+            spark_version: $DEFAULT_SPARK_VERSION
+            node_type_id: $NODE_TYPE_ID
+
+  sql_warehouses:
+    warehouse:
+      name: foo_$UNIQUE_NAME
+      cluster_size: Small

acceptance/bundle/resources/permissions/factcheck/out.test.toml

@@ -0,0 +1,33 @@
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test_bundle_[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+=== Permissions API simply takes the latest level in the request for a given principal
+jobs CAN_VIEW,CAN_MANAGE_RUN => CAN_MANAGE_RUN
+jobs CAN_MANAGE_RUN,CAN_VIEW => CAN_VIEW
+jobs CAN_VIEW,CAN_MANAGE_RUN,CAN_MANAGE => CAN_MANAGE
+jobs CAN_MANAGE,CAN_MANAGE_RUN => CAN_MANAGE_RUN
+jobs CAN_MANAGE,IS_OWNER => IS_OWNER
+
+=== Since we take the most recent level, IS_OWNER is lost, which results in the error due to lack of owner defined
+jobs IS_OWNER,CAN_MANAGE => SET ERROR Error: The job must have exactly one owner.
+REQUEST:
+{'access_control_list': [{'permission_level': 'IS_OWNER',
+                          'user_name': '[USERNAME]'},
+                         {'permission_level': 'CAN_MANAGE',
+                          'user_name': '[USERNAME]'}]}
+sql/warehouses CAN_VIEW,CAN_USE => CAN_USE
+sql/warehouses CAN_USE,CAN_VIEW => CAN_VIEW
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete job job
+  delete sql_warehouse warehouse
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test_bundle_[UNIQUE_NAME]/default
+
+Deleting files...
+Destroy complete!

acceptance/bundle/resources/permissions/factcheck/output.txt

@@ -0,0 +1,31 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+}
+trap cleanup EXIT
+
+
+trace $CLI bundle deploy
+
+job_id=$($CLI bundle summary --output json | jq -r '.resources.jobs.job.id')
+echo "$job_id:JOB_ID" >> ACC_REPLS
+
+sql_warehouse_id=$($CLI bundle summary --output json | jq -r '.resources.sql_warehouses.warehouse.id')
+echo "$sql_warehouse_id:SQL_WAREHOUSE_ID" >> ACC_REPLS
+
+[email protected]
+
+title "Permissions API simply takes the latest level in the request for a given principal\n"
+python3 check_permissions.py $user jobs $job_id CAN_VIEW,CAN_MANAGE_RUN CAN_MANAGE_RUN
+python3 check_permissions.py $user jobs $job_id CAN_MANAGE_RUN,CAN_VIEW CAN_VIEW
+python3 check_permissions.py $user jobs $job_id CAN_VIEW,CAN_MANAGE_RUN,CAN_MANAGE CAN_MANAGE
+python3 check_permissions.py $user jobs $job_id CAN_MANAGE,CAN_MANAGE_RUN CAN_MANAGE_RUN
+
+python3 check_permissions.py $CURRENT_USER_NAME jobs $job_id CAN_MANAGE,IS_OWNER IS_OWNER
+
+title "Since we take the most recent level, IS_OWNER is lost, which results in the error due to lack of owner defined\n"
+musterr python3 check_permissions.py $CURRENT_USER_NAME jobs $job_id IS_OWNER,CAN_MANAGE CAN_MANAGE
+
+python3 check_permissions.py $user sql/warehouses $sql_warehouse_id CAN_VIEW,CAN_USE CAN_USE
+python3 check_permissions.py $user sql/warehouses $sql_warehouse_id CAN_USE,CAN_VIEW CAN_VIEW

acceptance/bundle/resources/permissions/factcheck/script

@@ -0,0 +1,9 @@
+Local = false
+Cloud = true
+RecordRequests = false
+
+# I get this error, not sure why:
+#   -=== Since we take the most recent level, IS_OWNER is lost, which results in the error due to lack of owner defined
+#   -jobs IS_OWNER,CAN_MANAGE => SET ERROR Error: The job must have exactly one owner.
+#   +jobs CAN_VIEW,CAN_MANAGE_RUN => SET ERROR Error: [USERNAME]: Service[USERNAME]Name([USERNAME]) does not exist
+CloudEnvs.gcp = false