test(): Test policy to create and manage privileges (#9173)

smoke-test/tests/privileges/test_privileges.py

@@ -114,6 +114,21 @@ def _ensure_can_create_access_token(session, json):
     assert ingestion_data["data"]["createAccessToken"]["__typename"] == "AccessToken"
 
 
+@tenacity.retry(
+    stop=tenacity.stop_after_attempt(10), wait=tenacity.wait_fixed(sleep_sec)
+)
+def _ensure_can_create_user_policy(session, json):
+    response = session.post(f"{get_frontend_url()}/api/v2/graphql", json=json)
+    response.raise_for_status()
+    res_data = response.json()
+
+    assert res_data
+    assert res_data["data"]
+    assert res_data["data"]["createPolicy"] is not None
+
+    return res_data["data"]["createPolicy"] 
+
+
 @pytest.mark.dependency(depends=["test_healthchecks"])
 def test_privilege_to_create_and_manage_secrets():
 
@@ -337,4 +352,99 @@ def test_privilege_to_create_and_manage_access_tokens():
 
 
     # Ensure that user can't create access token after policy is removed
-    _ensure_cant_perform_action(user_session, create_access_token,"createAccessToken")
+    _ensure_cant_perform_action(user_session, create_access_token,"createAccessToken")
+
+
+@pytest.mark.dependency(depends=["test_healthchecks"])
+def test_privilege_to_create_and_manage_policies():
+
+    (admin_user, admin_pass) = get_admin_credentials()
+    admin_session = login_as(admin_user, admin_pass)
+    user_session = login_as("user", "user")
+
+
+    # Verify new user can't create a policy
+    create_policy = {
+        "query": """mutation createPolicy($input: PolicyUpdateInput!) {\n
+            createPolicy(input: $input) }""",
+        "variables": {
+            "input": {
+                "type": "PLATFORM",
+                "name": "Policy Name",
+                "description": "Policy Description",
+                "state": "ACTIVE",
+                "resources": {"filter":{"criteria":[]}},
+                "privileges": ["MANAGE_POLICIES"],
+                "actors": {
+                    "users": [],
+                    "resourceOwners": False,
+                    "allUsers": True,
+                    "allGroups": False,
+                },
+            }
+        },
+    }
+
+    _ensure_cant_perform_action(user_session, create_policy,"createPolicy")
+
+
+    # Assign privileges to the new user to create and manage policies
+    admin_policy_urn = create_user_policy("urn:li:corpuser:user", ["MANAGE_POLICIES"], admin_session)
+
+
+    # Verify new user can create and manage policy(create, edit, delete)
+    # Create a policy
+    user_policy_urn = _ensure_can_create_user_policy(user_session, create_policy)
+
+    # Edit a policy
+    edit_policy = { 
+        "query": """mutation updatePolicy($urn: String!, $input: PolicyUpdateInput!) {\n
+            updatePolicy(urn: $urn, input: $input) }""",
+        "variables": {
+            "urn": user_policy_urn,
+            "input": {
+                "type": "PLATFORM",
+                "state": "INACTIVE",
+                "name": "Policy Name test",
+                "description": "Policy Description updated",
+                "privileges": ["MANAGE_POLICIES"],
+                "actors": {
+                    "users": [],
+                    "groups": None,
+                    "resourceOwners": False,
+                    "allUsers": True,
+                    "allGroups": False,
+                    "resourceOwnersTypes": None,
+                },
+            },
+        },
+    }
+    edit_policy_response = user_session.post(f"{get_frontend_url()}/api/v2/graphql", json=edit_policy)
+    edit_policy_response.raise_for_status()
+    res_data = edit_policy_response.json()
+
+    assert res_data
+    assert res_data["data"]
+    assert res_data["data"]["updatePolicy"] == user_policy_urn
+
+    # Delete a policy
+    remove_user_policy = { 
+        "query": "mutation deletePolicy($urn: String!) {\n  deletePolicy(urn: $urn)\n}\n",
+        "variables":{"urn":user_policy_urn}
+    }
+
+    remove_policy_response = user_session.post(f"{get_frontend_url()}/api/v2/graphql", json=remove_user_policy)
+    remove_policy_response.raise_for_status()
+    res_data = remove_policy_response.json()
+
+    assert res_data
+    assert res_data["data"]
+    assert res_data["data"]["deletePolicy"] == user_policy_urn
+
+
+    # Remove the user privilege by admin
+    remove_policy(admin_policy_urn, admin_session)
+
+
+    # Ensure that user can't create a policy after privilege is removed by admin
+    _ensure_cant_perform_action(user_session, create_policy,"createPolicy")