planetstack/tests/useraccesstest.py

   1 import inspect
   2 import json
   3 import os
   4 import requests
   5 import sys
   6
   7 from operator import itemgetter, attrgetter
   8
   9 REST_API="http://node43.princeton.vicci.org:8000/plstackapi/"
  10 USERS_API = REST_API + "users/"
  11 SLICES_API = REST_API + "slices/"
  12 SITES_API = REST_API + "sites/"
  13 SITEPRIV_API = REST_API + "site_privileges/"
  14 SLICEPRIV_API = REST_API + "slice_memberships/"
  15 SITEROLE_API = REST_API + "site_roles/"
  16
  17 username = sys.argv[1]
  18 password = sys.argv[2]
  19
  20 opencloud_auth=(username, password)
  21 admin_auth=("scott@onlab.us", "letmein")
  22
  23 def fail_unless(x, msg):
  24     if not x:
  25         (frame, filename, line_number, function_name, lines, index) = inspect.getouterframes(inspect.currentframe())[1]
  26         print "FAIL (%s:%d)" % (function_name, line_number), msg
  27
  28
  29 print "downloading objects using admin"
  30 r = requests.get(USERS_API + "?no_hyperlinks=1", auth=admin_auth)
  31 allUsers = r.json()
  32 r = requests.get(SLICES_API + "?no_hyperlinks=1", auth=admin_auth)
  33 allSlices = r.json()
  34 r = requests.get(SITES_API + "?no_hyperlinks=1", auth=admin_auth)
  35 allSites = r.json()
  36 r = requests.get(SITEPRIV_API + "?no_hyperlinks=1", auth=admin_auth)
  37 allSitePriv = r.json()
  38 r = requests.get(SLICEPRIV_API + "?no_hyperlinks=1", auth=admin_auth)
  39 allSlicePriv = r.json()
  40 r = requests.get(SITEROLE_API + "?no_hyperlinks=1", auth=admin_auth)
  41 allSiteRole = r.json()
  42
  43 def should_see_user(myself, otherUser):
  44     if myself["is_admin"]:
  45         return True
  46     if myself["id"] == otherUser["id"]:
  47         return True
  48     for sitePriv in allSitePriv:
  49         if (sitePriv["user"] == myself["id"]) and (sitePriv["site"] == otherUser["site"]):
  50             for role in allSiteRole:
  51                 if role["role"]=="pi" and role["id"] == sitePriv["role"]:
  52                     return True
  53     return False
  54
  55 def flip_phone(user):
  56     if user["phone"] == "123":
  57         user["phone"] = "456"
  58     else:
  59         user["phone"] = "123"
  60
  61 print "  loaded user:%d slice:%d, site:%d, site_priv:%d slice_priv:%d" % (len(allUsers), len(allSlices), len(allSites), len(allSitePriv), len(allSlicePriv))
  62
  63 # get our own user record
  64
  65 r = requests.get(USERS_API + "?email=%s&no_hyperlinks" % username, auth=opencloud_auth)
  66 fail_unless(r.status_code==200, "failed to get user %s" % username)
  67 myself = r.json()
  68 fail_unless(len(myself)==1, "wrong number of results when getting user %s" % username)
  69 myself = myself[0]
  70
  71 # check to see that we see the users we should be able to
  72
  73 r = requests.get(USERS_API, auth=opencloud_auth)
  74 myUsers = r.json()
  75 for user in myUsers:
  76     fail_unless(should_see_user(myself, user), "saw user %s but we shouldn't have" % user["email"])
  77 myUsersIds = [r["id"] for r in myUsers]
  78 for user in allUsers:
  79     if should_see_user(myself, user):
  80         fail_unless(user["id"] in myUsersIds, "should have seen user %s but didnt" % user["email"])
  81
  82 # toggle the phone number on the users we should be able to
  83
  84 for user in allUsers:
  85     user = requests.get(USERS_API + str(user["id"]) + "/", auth=admin_auth).json()
  86     flip_phone(user)
  87     r = requests.put(USERS_API + str(user["id"]) +"/", data=user, auth=opencloud_auth)
  88     if should_see_user(myself, user):
  89         fail_unless(r.status_code==200, "failed to change phone number on %s" % user["email"])
  90     else:
  91         # XXX: this is failing, but for the wrong reason
  92         fail_unless(r.status_code!=200, "was able to change phone number on %s but shouldn't have" % user["email"])
  93
  94 for user in allUsers:
  95     user = requests.get(USERS_API + str(user["id"]) + "/", auth=admin_auth).json()
  96     user["is_staff"] = not user["is_staff"]
  97     r = requests.put(USERS_API + str(user["id"]) +"/", data=user, auth=opencloud_auth)
  98     if myself["is_admin"]:
  99         fail_unless(r.status_code==200, "failed to change is_staff on %s" % user["email"])
 100     else:
 101         # XXX: this is failing, but for the wrong reason
 102         fail_unless(r.status_code!=200, "was able to change is_staff on %s but shouldn't have" % user["email"])
 103
 104     # put it back to false, in case we successfully changed it...
 105     user["is_staff"] = False
 106     r = requests.put(USERS_API + str(user["id"]) +"/", data=user, auth=opencloud_auth)
 107
 108
 109
 110
 111