Allow running loading method twice

Author: AlanCoding

ansible_base/rbac/models/content_type.py

@@ -176,7 +176,8 @@ def get_for_id(self, id: int) -> django_models.Model:
 
     def load_remote_objects(self, remote_data: list[dict]):
         parent_mapping: dict[django_models.Model, str] = {}
-        for remote_type in remote_data:
+        for remote_type_raw in remote_data:
+            remote_type = remote_type_raw.copy()
             service = remote_type.pop('service')
             model = remote_type.pop('model')
             pct_slug = remote_type.pop('parent_content_type')
@@ -281,7 +282,12 @@ def model_class(self) -> Union[Type[django_models.Model], Type[RemoteObject]]:
 
             return get_remote_standin_class(self)
 
-        return apps.get_model(self.app_label, self.model)
+        try:
+            return apps.get_model(self.app_label, self.model)
+        except LookupError as exc:
+            raise LookupError(
+                f'Could not find ({self.app_label}, {self.model}), expected in local service={get_local_resource_prefix()} ' f'object service={self.service}'
+            ) from exc
 
     def get_object_for_this_type(self, **kwargs: Any) -> Union[django_models.Model, RemoteObject]:
         """Return the object referenced by this content type."""

ansible_base/rbac/models/permission.py

@@ -16,11 +16,12 @@ def load_remote_objects(self, remote_data: list[dict], update_managed=False):
         update_managed being True will refresh the managed roles like Organization Admin
         so that if the algorithm includes any new permissions, those are added.
         """
-        for remote_type in remote_data:
-            codename = remote_type.pop('codename')
-            ct_slug = remote_type.pop('content_type')
+        for remote_perm_raw in remote_data:
+            remote_perm = remote_perm_raw.copy()
+            codename = remote_perm.pop('codename')
+            ct_slug = remote_perm.pop('content_type')
             ct = DABContentType.objects.get(api_slug=ct_slug)
-            ct, _ = self.get_or_create(codename=codename, content_type=ct, defaults=remote_type)
+            ct, _ = self.get_or_create(codename=codename, content_type=ct, defaults=remote_perm)
 
         if update_managed:
             from ansible_base.rbac import permission_registry

test_app/tests/rbac/remote/test_managed_role_remote_perms.py

@@ -16,43 +16,51 @@ def test_remote_permissions_on_create(foo_permission):
 
 @pytest.mark.django_db
 def test_remote_permission_refresh(foo_type):
-    rd = RoleDefinition.objects.managed.org_admin
+    org_admin_rd = RoleDefinition.objects.managed.org_admin
+    auditor_rd = RoleDefinition.objects.managed.sys_auditor
 
     foo_permission = DABPermission.objects.create(codename='foo_foo', content_type=foo_type)
+    view_foo = DABPermission.objects.create(codename='view_foo', content_type=foo_type)
 
     # the foo_permission was created after the managed role
     # so the role is not explicitly created
-    assert foo_permission not in rd.permissions.all()
+    assert foo_permission not in org_admin_rd.permissions.all()
+    assert view_foo not in auditor_rd.permissions.all()
 
     # Loading it requires the refresh call
     permission_registry.create_managed_roles(apps, update_perms=True)
-    assert foo_permission in rd.permissions.all()
+    assert foo_permission in org_admin_rd.permissions.all()
+    assert view_foo in auditor_rd.permissions.all()
+
+    # calling multiple times should be okay
+    permission_registry.create_managed_roles(apps, update_perms=True)
 
 
 @pytest.mark.django_db
 def test_remote_permission_load_update_roles():
     rd = RoleDefinition.objects.managed.org_admin
-    DABContentType.objects.load_remote_objects(
-        [
-            {'service': 'fooland', 'app_label': 'foo', 'model': 'foo', 'api_slug': 'fooland.foo', 'parent_content_type': 'shared.organization'},
-            {'service': 'fooland', 'app_label': 'foo', 'model': 'bar', 'api_slug': 'fooland.bar', 'parent_content_type': None},
-        ]
-    )
+    remote_types = [
+        {'service': 'fooland', 'app_label': 'foo', 'model': 'foo', 'api_slug': 'fooland.foo', 'parent_content_type': 'shared.organization'},
+        {'service': 'fooland', 'app_label': 'foo', 'model': 'bar', 'api_slug': 'fooland.bar', 'parent_content_type': None},
+    ]
+    DABContentType.objects.load_remote_objects(remote_types)
     assert not DABPermission.objects.filter(codename='foo_foo').exists()
-    DABPermission.objects.load_remote_objects(
-        [
-            {'codename': 'foo_foo', 'content_type': 'fooland.foo', 'api_slug': "fooland.foo_foo"},
-            {'codename': 'bar_bar', 'content_type': 'fooland.bar', 'api_slug': "fooland.bar_bar"},
-        ],
-        update_managed=True,
-    )
+    remote_perms = [
+        {'codename': 'foo_foo', 'content_type': 'fooland.foo', 'api_slug': "fooland.foo_foo"},
+        {'codename': 'bar_bar', 'content_type': 'fooland.bar', 'api_slug': "fooland.bar_bar"},
+    ]
+    DABPermission.objects.load_remote_objects(remote_perms, update_managed=True)
     assert DABPermission.objects.filter(codename='foo_foo').exists()
 
     foo_permission = DABPermission.objects.get(codename='foo_foo')
     bar_permission = DABPermission.objects.get(codename='bar_bar')
     assert foo_permission in rd.permissions.all()
     assert bar_permission not in rd.permissions.all()  # not child type of organization
 
+    # calling twice should be fine
+    DABContentType.objects.load_remote_objects(remote_types)
+    DABPermission.objects.load_remote_objects(remote_perms, update_managed=True)
+
 
 @pytest.mark.django_db
 def test_remote_permission_duplicate_name():