Make assignment list fully function with remote permissions

add test Docstrings

Author: AlanCoding

ansible_base/rbac/models/content_type.py

@@ -221,6 +221,8 @@ def app_labeled_name(self) -> str:
         model = self.model_class()
         if not model:
             return self.model
+        if issubclass(model, RemoteObject):
+            return f'RemoteObject | {self.model}'
         return f"{model._meta.app_config.verbose_name} | {model._meta.verbose_name}"
 
     def save(self, *args, **kwargs):

ansible_base/rbac/models/fields.py

@@ -9,7 +9,7 @@
 from django.db.models.sql import AND
 from django.db.models.sql.where import WhereNode
 
-from ..remote import get_local_resource_prefix
+from ..remote import RemoteObject, get_local_resource_prefix
 from .content_type import DABContentType
 
 
@@ -59,6 +59,8 @@ def _check_content_type_field(self):
 
     def get_content_type(self, obj=None, id=None, using=None, model=None):
         if obj is not None:
+            if isinstance(obj, RemoteObject):
+                return obj.content_type
             return DABContentType.objects.db_manager(obj._state.db).get_for_model(
                 obj.__class__,
             )

ansible_base/rbac/remote.py

@@ -26,18 +26,45 @@ def __init__(self, ct: models.Model, abstract=False):
         self.app_label = ct.app_label
         self.abstract = abstract
 
+        class PK:
+            def __init__(self, pk_field_type):
+                self.pk_field_type = pk_field_type
+
+            def get_prep_value(self, value):
+                # TODO: handle uuid fields based on
+                # if self.pk_field_type:
+                return int(value)
+
+            def to_python(self, value):
+                return int(value)
+
+        self.pk = PK(ct.pk_field_type)
+
 
 class RemoteObject:
     """Placeholder for objects that live in another project."""
 
     def __init__(self, content_type: models.Model, object_id: Union[int, str]):
         self.content_type = content_type
         self.object_id = object_id
-        self._meta = StandinMeta(content_type, abstract=True)
+        if not hasattr(self, '_meta'):
+            # If object is created without a type-specific subclass, do the best we can
+            self._meta = StandinMeta(content_type, abstract=True)
+        else:
+            if content_type.model != self._meta.model_name:
+                raise RuntimeError(f'RemoteObject created with type {content_type} but with type for {self._meta.model_name}')
 
     def __repr__(self):
         return f"<RemoteObject {self.content_type} id={self.object_id}>"
 
+    def __eq__(self, value):
+        if isinstance(value, RemoteObject):
+            return bool(self.content_type.id == value.content_type.id and self.pk == value.pk)
+        return super().__eq__(value)
+
+    def __hash__(self):
+        return hash((self.content_type.id, self.pk))
+
     @classmethod
     def get_ct_from_type(cls):
         if not hasattr(cls, '_meta'):
@@ -59,7 +86,16 @@ def access_ids_qs(cls, actor, codename: str = 'view', content_types=None, cast_f
 
     @property
     def pk(self):
-        return self.object_id
+        """Alias to :attr:`object_id` for compatibility with Django. Also, handles type."""
+        return self._meta.pk.to_python(self.object_id)
+
+    def summary_fields(self):
+        """This gives a placeholder, planned to introduce a summary_fields shared endpoint.
+
+        This placeholder should be cleary identifable by a client or by the RBAC resource server.
+        Then, the idea, is that it can make a request to the remote server to get the summary data.
+        """
+        return {'<remote_object_placeholder>': True, 'model_name': self._meta.model_name, 'service': self._meta.service, 'pk': self.pk}
 
 
 def get_remote_base_class() -> Type[RemoteObject]:

test_app/tests/rbac/remote/test_public_api_compat.py

@@ -1,10 +1,12 @@
 import pytest
 
 from ansible_base.lib.utils.response import get_relative_url
+from ansible_base.rbac.remote import RemoteObject
 
 
 @pytest.mark.django_db
 def test_role_definition_list_remote_and_local(admin_api_client, inv_rd, foo_rd):
+    "Test that the role_definitions endpoint does not choke when remote permissions are listed."
     url = get_relative_url('roledefinition-list')
     response = admin_api_client.get(url)
     assert response.status_code == 200
@@ -18,10 +20,8 @@ def test_role_definition_list_remote_and_local(admin_api_client, inv_rd, foo_rd)
 
 
 @pytest.mark.django_db
-def test_create_remote_role_definition(admin_api_client, foo_type, foo_permission):
-    """
-    Test creation of a custom, remote role definition.
-    """
+def test_create_remote_role_definition_for_remote(admin_api_client, foo_type, foo_permission):
+    "Test creation of a custom role definition that gives permission to remote things."
     url = get_relative_url("roledefinition-list")
     data = dict(name='foo-foo-foo-custom', description='bar', permissions=[foo_permission.api_slug], content_type=foo_type.api_slug)
     response = admin_api_client.post(url, data=data, format="json")
@@ -30,26 +30,26 @@ def test_create_remote_role_definition(admin_api_client, foo_type, foo_permissio
     assert response.data['permissions'] == ['foo.foo_foo']
 
 
-# TODO: check that assignment endpoint works
-
-# @pytest.mark.django_db
-# def test_give_remote_permission(rando, foo_type, foo_permission, foo_rd):
-#     assert foo_type.service == 'foo'  # a place, a domain, a server, known as foo
-#     assert foo_type.api_slug == 'foo.foo'  # there lives a foo in foo
-
-#     assert foo_permission.api_slug == 'foo.foo_foo'  # expression of the ability that one may foo a foo
-
-#     a_foo = RemoteObject(content_type=foo_type, object_id=42)
-#     assignment = foo_rd.give_permission(rando, a_foo)
-
-#     assignment = RoleUserAssignment.objects.get(pk=assignment.pk)
-#     assert isinstance(assignment.content_object, RemoteObject)
-
-#     # We can do evaluation querysets, but these can not return objects, just id values
-#     assert set(foo_type.model_class().access_ids_qs(actor=rando, codename='foo')) == {(int(assignment.object_id),)}
-
-#     # Test that user-attached methods also work
-#     assert rando.has_obj_perm(a_foo, 'foo')
-#     with pytest.raises(RuntimeError) as exc:
-#         assert not rando.has_obj_perm(a_foo, 'bar')  # not a valid permission
-#     assert 'The permission bar_foo is not valid for model foo' in str(exc)
+@pytest.mark.django_db
+def test_user_role_assignment_remote_and_local(admin_api_client, rando, foo_type, foo_rd):
+    "Test that after assigning permission to remote objects the assignment list works."
+    a_foo = RemoteObject(content_type=foo_type, object_id=42)
+    assignment = foo_rd.give_permission(rando, a_foo)
+    assignment.content_object
+
+    assert isinstance(assignment.content_object, RemoteObject)
+
+    # Should show up in the assignments list
+    url = get_relative_url('roleuserassignment-list')
+    response = admin_api_client.get(url, format="json")
+    assert response.status_code == 200, response.data
+
+    data_by_rd = {item['role_definition']: item for item in response.data['results']}
+    assert foo_rd.id in data_by_rd
+    item = data_by_rd[foo_rd.id]
+    assert item['user'] == rando.id
+    assert item['object_id'] == str(a_foo.object_id)
+    assert 'summary_fields' in item
+    sf = item['summary_fields']
+    assert 'content_object' in sf
+    assert sf['content_object'] == {'<remote_object_placeholder>': True, 'model_name': 'foo', 'service': 'foo', 'pk': 42}

test_app/tests/rbac/remote/test_remote_assignment.py

@@ -2,20 +2,26 @@
 
 from ansible_base.rbac.models import RoleUserAssignment
 from ansible_base.rbac.remote import RemoteObject
+from test_app.models import User
 
 
 @pytest.mark.django_db
 def test_give_remote_permission(rando, foo_type, foo_permission, foo_rd):
+    "Test for a few technical elements related to giving permission to remote objects, and using in query."
     assert foo_type.service == 'foo'  # a place, a domain, a server, known as foo
     assert foo_type.api_slug == 'foo.foo'  # there lives a foo in foo
 
     assert foo_permission.api_slug == 'foo.foo_foo'  # expression of the ability that one may foo a foo
 
     a_foo = RemoteObject(content_type=foo_type, object_id=42)
     assignment = foo_rd.give_permission(rando, a_foo)
+    assert '42' in repr(a_foo)
+    assert a_foo.pk == 42
 
     assignment = RoleUserAssignment.objects.get(pk=assignment.pk)
+    assert assignment.content_object.pk == 42
     assert isinstance(assignment.content_object, RemoteObject)
+    assert isinstance(assignment.content_object, RemoteObject)  # There was a bug where multiple references would error
 
     # We can do evaluation querysets, but these can not return objects, just id values
     assert set(foo_type.model_class().access_ids_qs(actor=rando, codename='foo')) == {(int(assignment.object_id),)}
@@ -25,3 +31,18 @@ def test_give_remote_permission(rando, foo_type, foo_permission, foo_rd):
     with pytest.raises(RuntimeError) as exc:
         assert not rando.has_obj_perm(a_foo, 'bar')  # not a valid permission
     assert 'The permission bar_foo is not valid for model foo' in str(exc)
+
+
+@pytest.mark.django_db
+def test_prefetch_related_objects(foo_type, foo_rd, inv_rd, inventory):
+    users = [User.objects.create(username=f'user{i}') for i in range(10)]
+
+    a_foo = RemoteObject(content_type=foo_type, object_id=42)
+    for u in users:
+        foo_rd.give_permission(u, a_foo)
+        inv_rd.give_permission(u, inventory)
+
+    assert RoleUserAssignment.objects.count() == 20
+    assert {assignment.content_object for assignment in RoleUserAssignment.objects.all()} == {a_foo, inventory}
+    assert {assignment.content_object for assignment in RoleUserAssignment.objects.all()} == {a_foo, inventory}
+    assert {assignment.content_object for assignment in RoleUserAssignment.objects.prefetch_related('content_object')} == {a_foo, inventory}