Add methods to import types and permissions from API

Author: AlanCoding

ansible_base/rbac/models/content_type.py

@@ -165,6 +165,21 @@ def get_for_id(self, id: int) -> "DABContentType":
             self._add_to_cache(self.db, ct)
             return ct
 
+    def load_remote_types(self, remote_data: list[dict]):
+        parent_mapping: dict[django_models.Model, str] = {}
+        for remote_type in remote_data:
+            service = remote_type.pop('service')
+            model = remote_type.pop('model')
+            pct_slug = remote_type.pop('parent_content_type')
+            ct, _ = self.get_or_create(service=service, model=model, defaults=remote_type)
+            parent_mapping[ct] = pct_slug
+
+        # The parent type link needs to be filled in via a second pass
+        for ct, pct_slug in parent_mapping.items():
+            if pct_slug and (ct.parent_content_type_id is None or ct.parent_content_type.api_slug != pct_slug):
+                ct.parent_content_type = DABContentType.objects.get(api_slug=pct_slug)
+                ct.save()
+
 
 class DABContentType(django_models.Model):
     """Like Django ContentType model but scoped by service."""

ansible_base/rbac/models/permission.py

@@ -4,6 +4,15 @@
 from .content_type import DABContentType
 
 
+class DABPermissionManager(models.Manager):
+    def load_remote_types(self, remote_data: list[dict]):
+        for remote_type in remote_data:
+            codename = remote_type.pop('codename')
+            ct_slug = remote_type.pop('content_type')
+            ct = DABContentType.objects.get(api_slug=ct_slug)
+            ct, _ = self.get_or_create(codename=codename, content_type=ct, defaults=remote_type)
+
+
 class DABPermission(models.Model):
     "This is a minimal copy of auth.Permission for internal use"
 
@@ -33,6 +42,8 @@ class DABPermission(models.Model):
         help_text=_("String to use for references to this type from other models in the API."),
     )
 
+    objects = DABPermissionManager()
+
     class Meta:
         app_label = 'dab_rbac'
         verbose_name = "permission"

test_app/tests/rbac/remote/test_service_api.py

@@ -1,6 +1,9 @@
+from copy import deepcopy
+
 import pytest
 
 from ansible_base.lib.utils.response import get_relative_url
+from ansible_base.rbac.models import DABContentType, DABPermission
 
 
 @pytest.mark.django_db
@@ -51,3 +54,48 @@ def test_role_definition_listed_as_resource(admin_api_client, org_admin_rd):
     assert resource_data['content_type'] == 'shared.organization'
     assert 'permissions' in detail.data['additional_data']
     assert 'aap.add_inventory' in detail.data['additional_data']['permissions']
+
+
+@pytest.mark.django_db
+def test_reload_types(admin_api_client):
+    url = get_relative_url('dabcontenttype-list')
+    response = admin_api_client.get(url + '?page_size=200', format="json")
+    assert response.status_code == 200, response.data
+
+    type_list = response.data['results']
+    original = deepcopy(type_list)
+
+    DABContentType.objects.all().delete()  # Delete all types, see if we get them back
+
+    DABContentType.objects.load_remote_types(type_list)
+
+    response = admin_api_client.get(url + '?page_size=200', format="json")
+    assert response.status_code == 200, response.data
+
+    assert response.data['results'] == original
+
+
+@pytest.mark.django_db
+def test_load_child_of_org():
+    DABContentType.objects.load_remote_types([{'service': 'fooland', 'app_label': 'foop', 'model': 'fooser', 'parent_content_type': 'shared.organization'}])
+    ct = DABContentType.objects.get(api_slug='fooland.fooser')
+    assert ct.parent_content_type.app_label == 'test_app'  # proves connection to existing
+
+
+@pytest.mark.django_db
+def test_reload_permissions(admin_api_client):
+    url = get_relative_url('dabpermission-list')
+    response = admin_api_client.get(url + '?page_size=200', format="json")
+    assert response.status_code == 200, response.data
+
+    perm_list = response.data['results']
+    original = deepcopy(perm_list)
+
+    DABPermission.objects.all().delete()  # Delete all permissions, see if we get them back
+
+    DABPermission.objects.load_remote_types(perm_list)
+
+    response = admin_api_client.get(url + '?page_size=200', format="json")
+    assert response.status_code == 200, response.data
+
+    assert response.data['results'] == original