Automatically set DAB types to id of content type to make life easier

Author: AlanCoding

ansible_base/rbac/management/create_types.py

@@ -30,25 +30,38 @@ def create_DAB_contenttypes(
     comes from the permission registry.
     """
     DABContentType = apps.get_model("dab_rbac", "DABContentType")
+    ContentType = apps.get_model("contenttypes", "ContentType")
 
     content_types = get_local_DAB_contenttypes(using, DABContentType)
 
     # TODO: add api_slug field when added
-    cts = []
+    ct_data = []
     for model in permission_registry.all_registered_models:
         service = get_resource_prefix(model)
         if (service, model._meta.model_name) not in content_types:
             # The content type is not seen in existing entries, add to list for creation
-            cts.append(
-                DABContentType(
-                    service=service,
-                    app_label=model._meta.app_label,
-                    model=model._meta.model_name,
-                )
+            ct_item_data = dict(
+                service=service,
+                app_label=model._meta.app_label,
+                model=model._meta.model_name,
             )
-    if not cts:
+            # To make usage earier in a transitional period, we will set the content type
+            # of any new entries created here to the id of its corresponding ContentType
+            # from the actual contenttypes app, allowing many filters to work
+            real_ct = ContentType.objects.get_for_model(model)
+            if not DABContentType.objects.filter(id=real_ct.id).exists():
+                ct_item_data['id'] = real_ct.id
+            ct_data.append(ct_item_data)
+    if not ct_data:
         return
-    DABContentType.objects.using(using).bulk_create(cts)
+
+    # To make usage earier in a transitional period, we will set the content type
+    # of any new entries created here to the id of its corresponding ContentType
+    # from the actual contenttypes app, allowing many filters to work
+    cts = []
+    for ct_item_data in ct_data:
+        cts.append(DABContentType.objects.create(**ct_item_data))
+
     if verbosity >= 2:
         for ct in cts:
             logger.debug("Adding DAB content type " f"'{ct.service}:{ct.app_label} | {ct.model}'")

ansible_base/rbac/models/content_type.py

@@ -143,13 +143,15 @@ def get_by_natural_key(self, *args: str) -> "DABContentType":
         if len(args) == 2:
             service = get_local_resource_prefix()
             app_label, model = args
+            kwargs = {'service__in': [get_local_resource_prefix(), 'shared'], 'app_label': app_label, 'model': model}
         else:
             service, app_label, model = args
+            kwargs = {'service': service, 'app_label': app_label, 'model': model}
         key = (service, app_label, model)
         try:
             return self._cache[self.db][key]
         except KeyError:
-            ct = self.get(service=service, app_label=app_label, model=model)
+            ct = self.get(**kwargs)
             self._add_to_cache(self.db, ct)
             return ct
 
@@ -230,7 +232,7 @@ def get_object_for_this_type(self, **kwargs: Any) -> Union[django_models.Model,
         """Return the object referenced by this content type."""
         model = self.model_class()
 
-        from .fields import get_remote_base_class
+        from ..remote import get_remote_base_class
 
         remote_base = get_remote_base_class()
 
@@ -246,7 +248,7 @@ def get_all_objects_for_this_type(self, **kwargs: Any) -> Union[django_models.Qu
         """Return all objects referenced by this content type."""
         model = self.model_class()
 
-        from .fields import get_remote_base_class
+        from ..remote import get_remote_base_class
 
         remote_base = get_remote_base_class()
         if issubclass(model, remote_base):

ansible_base/rbac/models/role.py

@@ -1,6 +1,7 @@
 import logging
 from collections.abc import Iterable
 from typing import Optional, Type, Union
+from uuid import UUID
 
 # Django
 from django.conf import settings
@@ -524,7 +525,7 @@ def descendent_roles(self):
             descendents.update(set(target_team.has_roles.all()))
         return descendents
 
-    def expected_direct_permissions(self, types_prefetch=None) -> set[tuple[str, int, Union[int, str]]]:
+    def expected_direct_permissions(self, types_prefetch=None) -> set[tuple[str, int, Union[int, str, UUID]]]:
         """The expected permissions that holding this ObjectRole confers to the holder
 
         This is given in the form of tuples, which represent RoleEvaluation entries.

ansible_base/rbac/remote.py

@@ -1,6 +1,7 @@
 import inspect
 from typing import Type, Union
 
+from django.apps import apps
 from django.conf import settings
 from django.db import models
 from django.utils.module_loading import import_string
@@ -28,6 +29,13 @@ def __init__(self, content_type: models.Model, object_id: Union[int, str]):
     def __repr__(self):
         return f"<RemoteObject {self.content_type} id={self.object_id}>"
 
+    @classmethod
+    def get_ct_from_type(cls):
+        if not hasattr(cls, '_meta'):
+            raise ValueError('Generlized RemoteObject can not obtain content_type from its class')
+        ct_model = apps.get_model('dab_rbac', 'DABContentType')
+        return ct_model.objects.get_by_natural_key(cls._meta.service, cls._meta.app_label, cls._meta.model)
+
 
 def get_remote_base_class() -> Type[RemoteObject]:
     """Return the class which represents remote objects.
@@ -103,6 +111,7 @@ def get_remote_standin_class(content_type: models.Model) -> Type:
 
         class StandinMeta:
             def __init__(self, ct: models.Model):
+                self.service = ct.service
                 self.model_name = ct.model
                 self.app_label = ct.app_label
 

test_app/tests/rbac/models/test_content_type.py

@@ -1,4 +1,5 @@
 import pytest
+from django.contrib.contenttypes.models import ContentType
 from django.db import models
 from django.test import TestCase
 from django.test.utils import isolate_apps
@@ -8,6 +9,14 @@
 from test_app.models import Inventory, Organization
 
 
+@pytest.mark.django_db
+def test_migration_shadows_real_contenttype():
+    assert DABContentType.objects.count() > 0  # sanity
+    for dab_ct in DABContentType.objects.all():
+        ct = ContentType.objects.get_by_natural_key(dab_ct.app_label, dab_ct.model)
+        assert ct.id == dab_ct.id
+
+
 @pytest.mark.django_db
 def test_auto_create_content_type():
     DABContentType.objects.all().delete()

test_app/tests/rbac/remote/test_remote_models.py

@@ -8,9 +8,7 @@
 def test_give_remote_permission(rando):
     foo_type = DABContentType.objects.create(service='foo', model='foo', app_label='foo')
     assert foo_type.service == 'foo'
-    # TODO:
-    # assert foo_type.model_class() is not None
-    foo_foo = DABPermission.objects.create(codename='foo_foo', content_type=foo_type)
+    DABPermission.objects.create(codename='foo_foo', content_type=foo_type)
     rd = RoleDefinition.objects.create_from_permissions(name='Foo fooers for the foos in foo service', permissions=['foo.foo_foo'], content_type=foo_type)
     a_foo = RemoteObject(content_type=foo_type, object_id=42)
     rd.give_permission(rando, a_foo)