checkin

dev/scripts.community/check_namespaces_with_validated_data.py

@@ -0,0 +1,122 @@
+import gzip
+import json
+import os
+
+from django.contrib.auth import get_user_model
+
+from galaxy_ng.app.models import Namespace
+from galaxy_ng.app.api.v1.models import LegacyNamespace
+from galaxy_ng.app.api.v1.models import LegacyRole
+from galaxy_ng.app.utils.namespaces import generate_v3_namespace_from_attributes
+from galaxy_ng.app.utils import rbac
+from pulp_ansible.app.models import Collection
+
+
+User = get_user_model()
+
+CHECKMODE = True
+if os.environ.get('CHECK_MODE') == "0":
+    CHECKMODE = False
+
+
+def match_old_user_to_local_user(udata):
+    old_username = udata['galaxy_username']
+    found_user = User.objects.filter(username=old_username).first()
+    if found_user:
+        return found_user
+
+    if udata.get('github_login_new'):
+        found_user = User.objects.filter(username=udata['github_login_new']).first()
+        if found_user:
+            return found_user
+
+    if udata.get('github_login'):
+        found_user = User.objects.filter(username=udata['github_login']).first()
+        if found_user:
+            return found_user
+
+    github_login = udata['github_login']
+    if udata.get('github_login_new'):
+        github_login = udata['github_login_new']
+    if github_login is None:
+        github_login = udata['galaxy_username']
+
+    #if github_login is None:
+    #    import epdb; epdb.st()
+
+    print(f'FIX - create user {github_login}')
+    if not CHECKMODE:
+        user,_ = User.objects.get_or_create(username=github_login)
+        return user
+
+    #import epdb; epdb.st()
+    return None
+
+
+def do_check():
+
+    known_usernames = list(User.objects.values_list('username', flat=True))
+    known_usernames = dict((x, None) for x in known_usernames)
+
+    known_v1_names = list(LegacyNamespace.objects.values_list('name', flat=True))
+    known_v1_names = dict((x, None) for x in known_v1_names)
+
+    known_v3_names = list(Namespace.objects.values_list('name', flat=True))
+    known_v3_names = dict((x, None) for x in known_v3_names)
+
+    collection_namespaces = list(Collection.objects.values_list('namespace', flat=True))
+    collection_namespaces = sorted(set(collection_namespaces))
+    role_namespaces = list(LegacyRole.objects.values_list('namespace__name', flat=True))
+    role_namespaces = sorted(set(role_namespaces))
+
+    # compressed for size ...
+    fn = 'user_namespace_map_validated.json.gz'
+    with gzip.open(fn, 'rb') as gz_file:
+        raw = gz_file.read()
+    umap = json.loads(raw)
+
+    old_ns_owners = {}
+    for k, v in umap.items():
+        for ns_name in v.get('owned_namespaces', []):
+            if ns_name not in old_ns_owners:
+                old_ns_owners[ns_name] = []
+            old_ns_owners[ns_name].append(k)
+
+    for ns_name in sorted(list(old_ns_owners.keys())):
+        print('-' * 50)
+        print(ns_name)
+        print('-' * 50)
+
+        # optimize ...
+        if ns_name not in collection_namespaces and ns_name not in role_namespaces:
+            continue
+
+        # figure this out later ...
+        if ns_name not in known_v1_names and ns_name not in known_v3_names:
+            continue
+
+        # figure this out later ...
+        is_valid_v3_name = ns_name in known_v3_names or ns_name == generate_v3_namespace_from_attributes(username=ns_name)
+        if not is_valid_v3_name:
+            continue
+
+        v3_namespace = Namespace.objects.filter(name=ns_name).first()
+        if not v3_namespace:
+            continue
+
+        current_owners = rbac.get_v3_namespace_owners(v3_namespace)
+        old_owners = [match_old_user_to_local_user(umap[x]) for x in old_ns_owners[ns_name] if umap[x].get('github_login_verified')]
+        missing_owners = [x for x in old_owners if x and x not in current_owners]
+
+        # looks good?
+        if not missing_owners:
+            continue
+
+        #print(ns_name)
+        for missing_owner in missing_owners:
+            print(f'FIX - add {missing_owner} to v3:{v3_namespace} owners')
+            if not CHECKMODE:
+                rbac.add_user_to_v3_namespace(missing_owner, v3_namespace)
+
+
+do_check()

dev/scripts.community/check_validated_user_namespace_map.py

@@ -19,11 +19,22 @@ def do_check():
     if os.environ.get('CHECK_MODE') == "0":
         checkmode = False
 
+    known_usernames = list(User.objects.values_list('username', flat=True))
+    known_usernames = dict((x, None) for x in known_usernames)
+
+    known_v1_names = list(LegacyNamespace.objects.values_list('name', flat=True))
+    known_v1_names = dict((x, None) for x in known_v1_names)
+
+    known_v3_names = list(Namespace.objects.values_list('name', flat=True))
+    known_v3_names = dict((x, None) for x in known_v3_names)
+
+    # compressed for size ...
     fn = 'user_namespace_map_validated.json.gz'
     with gzip.open(fn, 'rb') as gz_file:
         raw = gz_file.read()
     umap = json.loads(raw)
 
+    # check each upstream user one by one ...
     uids = list(umap.keys())
     uids = sorted(uids, key=lambda x: int(x))
     for uid in uids:
@@ -39,6 +50,12 @@ def do_check():
             (old_data.get('gitub_login_new') and old_data.get('gitub_login_new') != old_data['galaxy_username']):
             continue
 
+        # skip any users that have no current namespaces on this system ...
+        # this should theoretically skip making namespaces with no content
+        found_namespaces = [x for x in old_data.get('owned_namespaces', []) if x in known_v1_names or x in known_v3_names]
+        if not found_namespaces:
+            continue
+
         galaxy_username = old_data['galaxy_username']
         github_login = old_data['github_login']
         github_id = old_data['github_id']