First full check script

This version emits the first fix-up script as its output.

tools/scripts/ig-hotfix/role_check.py

@@ -1,15 +1,73 @@
-from collections import Counter
+from collections import defaultdict
+import sys
+import textwrap
 
-from awx.main.models.ha import InstanceGroup
+from django.contrib.contenttypes.models import ContentType
+
+from awx.main.fields import ImplicitRoleField
 from awx.main.models.rbac import Role
 
 
-for ig in InstanceGroup.objects.order_by('id'):
-    for f in ('admin_role', 'use_role', 'read_role'):
-        r = getattr(ig, f, None)
-        print(f"id={ig.id} {f} // r.id={getattr(r, 'id', None)} {getattr(r, 'content_type', None)} {getattr(r, 'object_id', None)} {getattr(r, 'role_field', None)}")
+crosslinked = defaultdict(dict)
+orphaned_roles = []
 
 
-ct = ContentType.objects.get(app_label='main', model='instancegroup')
-for r in Role.objects.filter(content_type=ct).order_by('id'):
-    print(f"id={r.id} instancegroup {r.object_id} {r.role_field}")
+for ct in ContentType.objects.order_by('id'):
+    cls = ct.model_class()
+    if not any(isinstance(f, ImplicitRoleField) for f in cls._meta.fields):
+        continue
+    for obj in cls.objects.all():
+        for f in cls._meta.fields:
+            if not isinstance(f, ImplicitRoleField):
+                continue
+            r = getattr(obj, f.name, None)
+            if not r:
+                sys.stderr.write(f"{cls} id={obj.id} {f.name} does not have a Role object\n")
+                crosslinked[(ct.id, obj.id)][f.name] = None
+                continue
+            if r.content_object != obj:
+                sys.stderr.write(f"{cls.__name__} id={obj.id} {f.name} is pointing to a Role that is assigned to a different object: role.id={r.id} {r.content_type!r} {r.object_id} {r.role_field}\n")
+                crosslinked[(ct.id, obj.id)][f.name] = None
+                continue
+
+
+sys.stderr.write('===================================\n')
+for r in Role.objects.exclude(role_field__startswith='system_').order_by('id'):
+    if not r.content_object:
+        sys.stderr.write(f"Role id={r.id} is missing a valid content_object: {r.content_type!r} {r.object_id} {r.role_field}\n")
+        orphaned_roles.append(r.id)
+        continue
+    rev = getattr(r.content_object, r.role_field, None)
+    if not rev:
+        continue
+    if r.id != rev.id:
+        sys.stderr.write(f"Role id={r.id} {r.content_type!r} {r.object_id} {r.role_field} is pointing to an object using a different role: id={rev.id} {rev.content_type!r} {rev.object_id} {rev.role_field}\n")
+        crosslinked[(r.content_type_id, r.object_id)][r.role_field] = r.id
+        continue
+
+
+sys.stderr.write('===================================\n')
+
+
+print(f"""\
+from django.contrib.contenttypes.models import ContentType
+
+from awx.main.models.rbac import Role
+
+""")
+
+print("# Role objects that are assigned to objects that do not exist")
+for r in orphaned_roles:
+    print(f"Role.objects.filter(id={r}).delete()")
+
+
+print("\n")
+print("# Resource objects that are pointing to the wrong Role.  Some of these")
+print("# do not have corresponding Roles anywhere, so delete the foreign key.")
+print("# For those, new Roles will be constructed upon save.\n")
+for (ct, obj), kv in crosslinked.items():
+    print(f"ct = ContentType.objects.get(id={ct})")
+    print(f"obj = ct.get_object_for_this_type(id={obj})")
+    for f, val in kv.items():
+        print(f"setattr(obj, '{f}_id', {val})")
+    print(f"obj.save()\n")