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 apiB = arvados.api(host=j["arvados_api_hosts"][1], token=j["superuser_tokens"][1], insecure=True)
   9 apiC = arvados.api(host=j["arvados_api_hosts"][2], token=j["superuser_tokens"][2], insecure=True)
  10
  11 ###
  12 ### Check users on API server "A" (the LoginCluster) ###
  13 ###
  14
  15 users = apiA.users().list().execute()
  16
  17 assert len(users["items"]) == 11
  18
  19 by_username = {}
  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
  46 ###
  47 ### Check users on API server "B" (federation member) ###
  48 ###
  49 users = apiB.users().list().execute()
  50 assert len(users["items"]) == 11
  51
  52 for i in range(2, 9):
  53     found = False
  54     for u in users["items"]:
  55         if (u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i) and
  56             u["uuid"] == by_username[u["username"]] and u["is_active"] is True):
  57             found = True
  58     assert found, "Not found case%i" % i
  59
  60 found = False
  61 for u in users["items"]:
  62     if (u["username"] == "case9" and u["email"] == "case9@test" and
  63         u["uuid"] == by_username[u["username"]] and u["is_active"] is False):
  64         found = True
  65 assert found
  66
  67
  68 ###
  69 ### Check users on API server "C" (federation member) ###
  70 ###
  71 users = apiC.users().list().execute()
  72 assert len(users["items"]) == 8
  73
  74 for i in (2, 4, 6, 7, 8):
  75     found = False
  76     for u in users["items"]:
  77         if (u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i) and
  78             u["uuid"] == by_username[u["username"]] and u["is_active"] is True):
  79             found = True
  80     assert found
  81
  82 # cases 3, 5, 9 involve users that have never accessed cluster C so
  83 # there's nothing to migrate.
  84 for i in (3, 5, 9):
  85     found = False
  86     for u in users["items"]:
  87         if (u["username"] == ("case%d" % i) and u["email"] == ("case%d@test" % i) and
  88             u["uuid"] == by_username[u["username"]] and u["is_active"] is True):
  89             found = True
  90     assert not found
  91
  92 print("Passed checks")