sdk/python/tests/fed-migrate/check.py

   1 import arvados
   2 import json
   3 import sys
   4
   5 j = json.load(open(sys.argv[1]))
   6
   7 apiA = arvados.api(host=j["arvados_api_hosts"][0], token=j["superuser_tokens"][0], insecure=True)
   8 tok = apiA.api_client_authorizations().current().execute()
   9 v2_token = "v2/%s/%s" % (tok["uuid"], tok["api_token"])
  10
  11 apiB = arvados.api(host=j["arvados_api_hosts"][1], token=v2_token, insecure=True)
  12 apiC = arvados.api(host=j["arvados_api_hosts"][2], token=v2_token, insecure=True)
  13
  14 ###
  15 ### Check users on API server "A" (the LoginCluster) ###
  16 ###
  17 by_username = {}
  18 def check_A(users):
  19     assert len(users["items"]) == 11
  20
  21     for i in range(1, 10):
  22         found = False
  23         for u in users["items"]:
  24             if u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i):
  25                 found = True
  26                 by_username[u["username"]] = u["uuid"]
  27         assert found
  28
  29     # Should be active
  30     for i in (1, 2, 3, 4, 5, 6, 7, 8):
  31         found = False
  32         for u in users["items"]:
  33             if u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i) and u["is_active"] is True:
  34                 found = True
  35         assert found, "Not found case%i" % i
  36
  37     # case9 should not be active
  38     found = False
  39     for u in users["items"]:
  40         if (u["username"] == "case9" and u["email"] == "case9@test" and
  41             u["uuid"] == by_username[u["username"]] and u["is_active"] is False):
  42             found = True
  43     assert found
  44
  45 users = apiA.users().list().execute()
  46 check_A(users)
  47
  48 users = apiA.users().list(bypass_federation=True).execute()
  49 check_A(users)
  50
  51 ###
  52 ### Check users on API server "B" (federation member) ###
  53 ###
  54
  55 # check for expected migrations on B
  56 users = apiB.users().list(bypass_federation=True).execute()
  57 assert len(users["items"]) == 11
  58
  59 for i in range(2, 9):
  60     found = False
  61     for u in users["items"]:
  62         if (u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i) and
  63             u["uuid"] == by_username[u["username"]] and u["is_active"] is True):
  64             found = True
  65     assert found, "Not found case%i" % i
  66
  67 found = False
  68 for u in users["items"]:
  69     if (u["username"] == "case9" and u["email"] == "case9@test" and
  70         u["uuid"] == by_username[u["username"]] and u["is_active"] is False):
  71         found = True
  72 assert found
  73
  74 # check that federated user listing works
  75 users = apiB.users().list().execute()
  76 check_A(users)
  77
  78 ###
  79 ### Check users on API server "C" (federation member) ###
  80 ###
  81
  82 # check for expected migrations on C
  83 users = apiC.users().list(bypass_federation=True).execute()
  84 assert len(users["items"]) == 8
  85
  86 for i in (2, 4, 6, 7, 8):
  87     found = False
  88     for u in users["items"]:
  89         if (u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i) and
  90             u["uuid"] == by_username[u["username"]] and u["is_active"] is True):
  91             found = True
  92     assert found
  93
  94 # cases 3, 5, 9 involve users that have never accessed cluster C so
  95 # there's nothing to migrate.
  96 for i in (3, 5, 9):
  97     found = False
  98     for u in users["items"]:
  99         if (u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i) and
 100             u["uuid"] == by_username[u["username"]] and u["is_active"] is True):
 101             found = True
 102     assert not found
 103
 104 # check that federated user listing works
 105 users = apiC.users().list().execute()
 106 check_A(users)
 107
 108 print("Passed checks")