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

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