Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 1 | """ useraccesstest.py |
| 2 | |
| 3 | This is a basic REST API permission test. Call it with a username and a |
| 4 | password, and it will try to read and update some user and slice object, |
| 5 | and report if something is broken. |
| 6 | |
| 7 | This is not an exhaustive test. |
| 8 | """ |
| 9 | |
| 10 | |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 11 | import inspect |
| 12 | import json |
| 13 | import os |
| 14 | import requests |
| 15 | import sys |
| 16 | |
| 17 | from operator import itemgetter, attrgetter |
| 18 | |
| 19 | REST_API="http://node43.princeton.vicci.org:8000/plstackapi/" |
| 20 | USERS_API = REST_API + "users/" |
| 21 | SLICES_API = REST_API + "slices/" |
| 22 | SITES_API = REST_API + "sites/" |
| 23 | SITEPRIV_API = REST_API + "site_privileges/" |
| 24 | SLICEPRIV_API = REST_API + "slice_memberships/" |
| 25 | SITEROLE_API = REST_API + "site_roles/" |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 26 | SLICEROLE_API = REST_API + "slice_roles/" |
| 27 | |
| 28 | TEST_USER_EMAIL = "test1234@test.com" |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 29 | |
| 30 | username = sys.argv[1] |
| 31 | password = sys.argv[2] |
| 32 | |
| 33 | opencloud_auth=(username, password) |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 34 | admin_auth=("scott@onlab.us", "letmein") # admin creds, used to get full set of objects |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 35 | |
| 36 | def fail_unless(x, msg): |
| 37 | if not x: |
| 38 | (frame, filename, line_number, function_name, lines, index) = inspect.getouterframes(inspect.currentframe())[1] |
| 39 | print "FAIL (%s:%d)" % (function_name, line_number), msg |
| 40 | |
| 41 | |
| 42 | print "downloading objects using admin" |
| 43 | r = requests.get(USERS_API + "?no_hyperlinks=1", auth=admin_auth) |
| 44 | allUsers = r.json() |
| 45 | r = requests.get(SLICES_API + "?no_hyperlinks=1", auth=admin_auth) |
| 46 | allSlices = r.json() |
| 47 | r = requests.get(SITES_API + "?no_hyperlinks=1", auth=admin_auth) |
| 48 | allSites = r.json() |
| 49 | r = requests.get(SITEPRIV_API + "?no_hyperlinks=1", auth=admin_auth) |
| 50 | allSitePriv = r.json() |
| 51 | r = requests.get(SLICEPRIV_API + "?no_hyperlinks=1", auth=admin_auth) |
| 52 | allSlicePriv = r.json() |
| 53 | r = requests.get(SITEROLE_API + "?no_hyperlinks=1", auth=admin_auth) |
| 54 | allSiteRole = r.json() |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 55 | r = requests.get(SLICEROLE_API + "?no_hyperlinks=1", auth=admin_auth) |
| 56 | allSliceRole = r.json() |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 57 | |
| 58 | def should_see_user(myself, otherUser): |
| 59 | if myself["is_admin"]: |
| 60 | return True |
| 61 | if myself["id"] == otherUser["id"]: |
| 62 | return True |
| 63 | for sitePriv in allSitePriv: |
| 64 | if (sitePriv["user"] == myself["id"]) and (sitePriv["site"] == otherUser["site"]): |
| 65 | for role in allSiteRole: |
| 66 | if role["role"]=="pi" and role["id"] == sitePriv["role"]: |
| 67 | return True |
| 68 | return False |
| 69 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 70 | def should_see_slice(myself, slice): |
| 71 | if myself["is_admin"]: |
| 72 | return True |
| 73 | for sitePriv in allSitePriv: |
| 74 | if (sitePriv["user"] == myself["id"]) and (sitePriv["site"] == slice["site"]): |
| 75 | for role in allSiteRole: |
| 76 | if role["role"]=="pi" and role["id"] == sitePriv["role"]: |
| 77 | return True |
| 78 | for slicePriv in allSlicePriv: |
| 79 | if (slicePriv["user"] == myself["id"]) and (sitePriv["slice"] == slice["id"]): |
| 80 | for role in allSliceRole: |
| 81 | if role["role"]=="pi" and role["id"] == slicePriv["role"]: |
| 82 | return True |
| 83 | return False |
| 84 | |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 85 | def flip_phone(user): |
| 86 | if user["phone"] == "123": |
| 87 | user["phone"] = "456" |
| 88 | else: |
| 89 | user["phone"] = "123" |
| 90 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 91 | def flip_desc(slice): |
| 92 | if slice["description"] == "some_description": |
| 93 | slice["description"] = "some_other_description" |
| 94 | else: |
| 95 | slice["description"] = "some_description" |
| 96 | |
| 97 | def delete_user_if_exists(email): |
| 98 | r = requests.get(USERS_API +"?email=%s" % email, auth=admin_auth) |
| 99 | if r.status_code==200: |
| 100 | user = r.json() |
| 101 | if len(user)>0: |
| 102 | user=user[0] |
| 103 | r = requests.delete(USERS_API + str(user["id"]) + "/", auth=admin_auth) |
| 104 | fail_unless(r.status_code==200, "failed to delete the test user") |
| 105 | |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 106 | print " loaded user:%d slice:%d, site:%d, site_priv:%d slice_priv:%d" % (len(allUsers), len(allSlices), len(allSites), len(allSitePriv), len(allSlicePriv)) |
| 107 | |
| 108 | # get our own user record |
| 109 | |
| 110 | r = requests.get(USERS_API + "?email=%s&no_hyperlinks" % username, auth=opencloud_auth) |
| 111 | fail_unless(r.status_code==200, "failed to get user %s" % username) |
| 112 | myself = r.json() |
| 113 | fail_unless(len(myself)==1, "wrong number of results when getting user %s" % username) |
| 114 | myself = myself[0] |
| 115 | |
| 116 | # check to see that we see the users we should be able to |
| 117 | |
| 118 | r = requests.get(USERS_API, auth=opencloud_auth) |
| 119 | myUsers = r.json() |
| 120 | for user in myUsers: |
| 121 | fail_unless(should_see_user(myself, user), "saw user %s but we shouldn't have" % user["email"]) |
| 122 | myUsersIds = [r["id"] for r in myUsers] |
| 123 | for user in allUsers: |
| 124 | if should_see_user(myself, user): |
| 125 | fail_unless(user["id"] in myUsersIds, "should have seen user %s but didnt" % user["email"]) |
| 126 | |
| 127 | # toggle the phone number on the users we should be able to |
| 128 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 129 | """ |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 130 | for user in allUsers: |
| 131 | user = requests.get(USERS_API + str(user["id"]) + "/", auth=admin_auth).json() |
| 132 | flip_phone(user) |
| 133 | r = requests.put(USERS_API + str(user["id"]) +"/", data=user, auth=opencloud_auth) |
| 134 | if should_see_user(myself, user): |
| 135 | fail_unless(r.status_code==200, "failed to change phone number on %s" % user["email"]) |
| 136 | else: |
| 137 | # XXX: this is failing, but for the wrong reason |
| 138 | fail_unless(r.status_code!=200, "was able to change phone number on %s but shouldn't have" % user["email"]) |
| 139 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 140 | # try changing is_staff. We should be able to do it if we're an admin, but not |
| 141 | # otherwise. |
| 142 | |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 143 | for user in allUsers: |
| 144 | user = requests.get(USERS_API + str(user["id"]) + "/", auth=admin_auth).json() |
| 145 | user["is_staff"] = not user["is_staff"] |
| 146 | r = requests.put(USERS_API + str(user["id"]) +"/", data=user, auth=opencloud_auth) |
| 147 | if myself["is_admin"]: |
| 148 | fail_unless(r.status_code==200, "failed to change is_staff on %s" % user["email"]) |
| 149 | else: |
| 150 | # XXX: this is failing, but for the wrong reason |
| 151 | fail_unless(r.status_code!=200, "was able to change is_staff on %s but shouldn't have" % user["email"]) |
| 152 | |
| 153 | # put it back to false, in case we successfully changed it... |
| 154 | user["is_staff"] = False |
| 155 | r = requests.put(USERS_API + str(user["id"]) +"/", data=user, auth=opencloud_auth) |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 156 | """ |
| 157 | |
| 158 | # delete the TEST_USER_EMAIL if it exists |
| 159 | delete_user_if_exists(TEST_USER_EMAIL) |
| 160 | |
| 161 | newUser = {"firstname": "test", "lastname": "1234", "email": TEST_USER_EMAIL, "password": "letmein"} |
| 162 | r = requests.post(USERS_API, data=newUser, auth=opencloud_auth) |
| 163 | if myself["is_admin"]: |
| 164 | fail_unless(r.status_code==200, "failed to create %s" % TEST_USER_EMAIL) |
| 165 | else: |
| 166 | fail_unless(r.status_code!=200, "created %s but we shouldn't have been able to" % TEST_USER_EMAIL) |
| 167 | |
| 168 | delete_user_if_exists(TEST_USER_EMAIL) |
| 169 | |
| 170 | sys.exit(-1) |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 171 | |
| 172 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 173 | # now create it as admin |
| 174 | r = requests.post(USERS_API, data=newUser, auth=admin_auth) |
| 175 | fail_unless(r.status_code==201, "failed to create %s as admin" % TEST_USER_EMAIL) |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 176 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 177 | user = requests.get(USERS_API +"?email=%s" % TEST_USER_EMAIL, auth=admin_auth).json()[0] |
| 178 | r = requests.delete(USERS_API + str(user["id"]) + "/", auth=opencloud_auth) |
| 179 | if myself["is_admin"]: |
| 180 | fail_unless(r.status_code==200, "failed to delete %s" % TEST_USER_EMAIL) |
| 181 | else: |
| 182 | fail_unless(r.status_code!=200, "deleted %s but we shouldn't have been able to" % TEST_USER_EMAIL) |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 183 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 184 | # slice tests |
Scott Baker | 51d4436 | 2014-10-07 12:54:51 -0700 | [diff] [blame] | 185 | |
Scott Baker | e282eae | 2014-10-09 10:59:58 -0700 | [diff] [blame^] | 186 | r = requests.get(SLICES_API, auth=opencloud_auth) |
| 187 | mySlices = r.json() |
| 188 | |
| 189 | for slice in mySlices: |
| 190 | fail_unless(should_see_slice(myself, slice), "saw slice %s but we shouldn't have" % slice["name"]) |
| 191 | mySlicesIds = [r["id"] for r in mySlices] |
| 192 | for slice in allSlices: |
| 193 | if should_see_slice(myself, slice): |
| 194 | fail_unless(slice["id"] in mySliceIds, "should have seen slice %s but didnt" % slice["name"]) |
| 195 | |
| 196 | for slice in allSlices: |
| 197 | slice = requests.get(SLICES_API + str(slice["id"]) + "/", auth=admin_auth).json() |
| 198 | flip_desc(slice) |
| 199 | r = requests.put(SLICES_API + str(slice["id"]) +"/", data=slice, auth=opencloud_auth) |
| 200 | if should_see_slice(myself, slice): |
| 201 | fail_unless(r.status_code==200, "failed to change desc on %s" % slice["name"]) |
| 202 | else: |
| 203 | fail_unless(r.status_code!=200, "was able to change desc on %s but shouldn't have" % slice["name"]) |
| 204 | |
| 205 | print "Done." |