Use api_slug in serializers

Author: AlanCoding

ansible_base/rbac/api/serializers.py

@@ -2,11 +2,9 @@
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import transaction
 from django.db.utils import IntegrityError
-from django.utils.functional import cached_property
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 from rest_framework.exceptions import PermissionDenied
-from rest_framework.fields import flatten_choices_dict, to_choices_dict
 from rest_framework.serializers import ValidationError
 
 from ansible_base.lib.abstract_models.common import get_url_for_object
@@ -16,117 +14,29 @@
 from ansible_base.rbac.policies import check_content_obj_permission, visible_users
 from ansible_base.rbac.validators import check_locally_managed, validate_permissions_for_model
 
-from ..remote import get_resource_prefix
-
-
-class ChoiceLikeMixin(serializers.ChoiceField):
-    """
-    This uses a ForeignKey to populate the choices of a choice field.
-    This also manages some string manipulation, right now, adding the local service name.
-    """
-
-    default_error_messages = serializers.PrimaryKeyRelatedField.default_error_messages
-
-    def get_dynamic_choices(self):
-        raise NotImplementedError
-
-    def get_dynamic_object(self, data):
-        raise NotImplementedError
-
-    def to_representation(self, value):
-        raise NotImplementedError
-
-    def __init__(self, **kwargs):
-        # Workaround so that the parent class does not resolve the choices right away
-        self.html_cutoff = kwargs.pop('html_cutoff', self.html_cutoff)
-        self.html_cutoff_text = kwargs.pop('html_cutoff_text', self.html_cutoff_text)
-
-        self.allow_blank = kwargs.pop('allow_blank', False)
-        super(serializers.ChoiceField, self).__init__(**kwargs)
-
-    def _initialize_choices(self):
-        choices = self.get_dynamic_choices()
-        self._grouped_choices = to_choices_dict(choices)
-        self._choices = flatten_choices_dict(self._grouped_choices)
-        self.choice_strings_to_values = {str(k): k for k in self._choices}
-
-    @cached_property
-    def grouped_choices(self):
-        self._initialize_choices()
-        return self._grouped_choices
-
-    @cached_property
-    def choices(self):
-        self._initialize_choices()
-        return self._choices
-
-    def to_internal_value(self, data):
-        try:
-            return self.get_dynamic_object(data)
-        except ObjectDoesNotExist:
-            self.fail('does_not_exist', pk_value=data)
-        except (TypeError, ValueError):
-            self.fail('incorrect_type', data_type=type(data).__name__)
-
-
-class ContentTypeField(ChoiceLikeMixin):
-    def __init__(self, **kwargs):
-        kwargs['help_text'] = _('The type of resource this applies to.')
-        super().__init__(**kwargs)
-
-    def get_resource_type_name(self, cls) -> str:
-        return f"{get_resource_prefix(cls)}.{cls._meta.model_name}"
-
-    def get_dynamic_choices(self):
-        return list(sorted((self.get_resource_type_name(cls), cls._meta.verbose_name.title()) for cls in permission_registry.all_registered_models))
-
-    def get_dynamic_object(self, data):
-        model = data.rsplit('.')[-1]
-        cls = permission_registry.get_model_by_name(model)
-        if cls is None:
-            return permission_registry.content_type_model.objects.none().get()  # raises correct DoesNotExist
-        return permission_registry.content_type_model.objects.get_for_model(cls)
-
-    def to_representation(self, value):
-        if isinstance(value, str):
-            return value  # slight hack to work to AWX schema tests
-        return self.get_resource_type_name(value.model_class())
-
-
-class PermissionField(ChoiceLikeMixin):
-    def get_dynamic_choices(self):
-        perms = []
-        for cls in permission_registry.all_registered_models:
-            cls_name = cls._meta.model_name
-            for action in cls._meta.default_permissions:
-                perms.append(f'{get_resource_prefix(cls)}.{action}_{cls_name}')
-            for perm_name, description in cls._meta.permissions:
-                perms.append(f'{get_resource_prefix(cls)}.{perm_name}')
-        return list(sorted(perms))
-
-    def get_dynamic_object(self, data):
-        codename = data.rsplit('.')[-1]
-        return permission_registry.permission_qs.get(codename=codename)
-
-    def to_representation(self, value):
-        if isinstance(value, str):
-            return value  # slight hack to work to AWX schema tests
-        ct = permission_registry.content_type_model.objects.get_for_id(value.content_type_id)  # optimization
-        return f'{get_resource_prefix(ct.model_class())}.{value.codename}'
-
-
-class ManyRelatedListField(serializers.ListField):
-    def to_representation(self, data):
-        "Adds the .all() to treat the value as a queryset"
-        return [self.child.to_representation(item) if item is not None else None for item in data.all()]
+from ..models import DABContentType, DABPermission
 
 
 class RoleDefinitionSerializer(CommonModelSerializer):
-    # Relational versions - we may switch to these if custom permission and type models are exposed but out of scope here
-    # permissions = serializers.SlugRelatedField(many=True, slug_field='codename', queryset=DABPermission.objects.all())
-    # content_type = ContentTypeField(slug_field='model', queryset=permission_registry.content_type_model.objects.all(), allow_null=True, default=None)
-    permissions = ManyRelatedListField(child=PermissionField())
-    content_type = ContentTypeField(allow_null=True, default=None)
+    permissions = serializers.SlugRelatedField(
+        slug_field='api_slug',
+        queryset=DABPermission.objects.all(),
+        many=True,
+        error_messages={
+            'does_not_exist': "Cannot use permission with api_slug '{value}', object does not exist",
+            'invalid': "Each content type must be a valid slug string",
+        },
+    )
+    content_type = serializers.SlugRelatedField(
+        slug_field='api_slug',
+        queryset=DABContentType.objects.all(),
+        allow_null=True,  # for global roles
+        default=None,
+        error_messages={
+            'does_not_exist': "Cannot use type with api_slug '{value}', object does not exist",
+            'invalid': "Each content type must be a valid slug string",
+        },
+    )
 
     class Meta:
         model = RoleDefinition
@@ -141,7 +51,7 @@ def validate(self, validated_data):
             permissions = list(self.instance.permissions.all())
         if 'content_type' in validated_data:
             content_type = validated_data['content_type']
-        else:
+        elif self.instance:
             content_type = self.instance.content_type
         validate_permissions_for_model(permissions, content_type)
         if getattr(self, 'instance', None):
@@ -150,11 +60,11 @@ def validate(self, validated_data):
 
 
 class RoleDefinitionDetailSerializer(RoleDefinitionSerializer):
-    content_type = ContentTypeField(read_only=True)
+    content_type = serializers.SlugRelatedField(slug_field='api_slug', read_only=True)
 
 
 class BaseAssignmentSerializer(CommonModelSerializer):
-    content_type = ContentTypeField(read_only=True)
+    content_type = serializers.SlugRelatedField(slug_field='api_slug', read_only=True)
     object_ansible_id = serializers.UUIDField(
         required=False,
         help_text=_('The resource id of the object this role applies to. An alternative to the object_id field.'),

test_app/tests/rbac/api/test_rbac_fields.py

@@ -1,13 +1,12 @@
 import pytest
 
-from ansible_base.lib.utils.response import get_relative_url
 from ansible_base.rbac.api.serializers import RoleDefinitionSerializer
 
 
 @pytest.mark.django_db
 def test_invalid_content_type(admin_api_client):
     serializer = RoleDefinitionSerializer(
-        data=dict(name='foo-role-def', description='bar', permissions=['aap.view_organization'], content_type='aap.foo_does_not_exist_model')
+        data=dict(name='foo-role-def', description='bar', permissions=['shared.view_organization'], content_type='aap.foo_does_not_exist_model')
     )
     assert not serializer.is_valid()
     assert 'object does not exist' in str(serializer.errors['content_type'])
@@ -22,18 +21,3 @@ def test_invalid_permission(admin_api_client):
     assert not serializer.is_valid()
     assert 'object does not exist' in str(serializer.errors['permissions'])
     assert 'content_type' not in serializer.errors
-
-
-@pytest.mark.django_db
-def test_parity_with_resource_registry(admin_api_client):
-    types_resp = admin_api_client.get(get_relative_url("resourcetype-list"))
-    assert types_resp.status_code == 200
-    res_types = set(r['name'] for r in types_resp.data['results'])
-
-    role_types = admin_api_client.options(get_relative_url("roledefinition-list"))
-    role_types = set(item['value'] for item in role_types.data['actions']['POST']['content_type']['choices'])
-
-    # Check the types in both registries
-    for type_name in ('shared.organization', 'shared.team'):
-        assert type_name in res_types
-        assert type_name in role_types

test_app/tests/rbac/api/test_rbac_validation.py

@@ -48,8 +48,8 @@ def test_custom_roles_for_shared_stuff_not_allowed(self, admin_api_client, allow
                 url,
                 data={
                     'name': 'Alternative Organization Admin Role in Local Server',
-                    'content_type': 'aap.organization',
-                    'permissions': ['aap.view_organization', 'local.change_organization'],
+                    'content_type': 'shared.organization',
+                    'permissions': ['shared.view_organization', 'shared.change_organization'],
                 },
             )
         if allowed is False:
@@ -77,8 +77,8 @@ def test_org_resource_roles_creatable(self, admin_api_client):
             url,
             data={
                 'name': 'Custom Organization Inventory Admin Role',
-                'content_type': 'aap.organization',
-                'permissions': ['aap.view_organization', 'local.change_inventory', 'local.view_inventory'],
+                'content_type': 'shared.organization',
+                'permissions': ['shared.view_organization', 'aap.change_inventory', 'aap.view_inventory'],
             },
         )
         assert response.status_code == 201, response.data
@@ -186,22 +186,3 @@ def test_callback_validate_role_team_assignment(admin_api_client, inventory, org
     response = admin_api_client.post(url, data={'object_id': inventory.id, 'team': team.id, 'role_definition': inv_rd.id})
     assert response.status_code == 403
     assert "Role assignment not allowed 403" in str(response.data)
-
-
-@pytest.mark.django_db
-def test_unregistered_model_type_for_assignment(admin_api_client, inventory, inv_rd, user):
-    url = get_relative_url('roleuserassignment-list')
-    # force invalid state of role definition, should not really happen
-    cls = apps.get_model('test_app.secretcolor')
-    inv_rd.content_type = permission_registry.content_type_model.objects.get_for_model(cls)
-    inv_rd.save(update_fields=['content_type'])
-    r = admin_api_client.post(url, data={'object_id': inventory.pk, 'user': user.pk, 'role_definition': inv_rd.pk})
-    assert r.status_code == 400
-    assert 'Given role definition is for a model not registered in the permissions system' in str(r.data)
-
-    # Temporarily abusing user model to test this via ansible_id reference, this testing method may not work forever
-    inv_rd.content_type = permission_registry.content_type_model.objects.get_for_model(user)
-    inv_rd.save(update_fields=['content_type'])
-    r = admin_api_client.post(url, data={'object_ansible_id': str(user.resource.ansible_id), 'user': user.pk, 'role_definition': inv_rd.pk})
-    assert r.status_code == 400
-    assert 'user type is not registered with DAB RBAC' in str(r.data)

test_app/tests/rbac/api/test_rbac_views.py

@@ -19,7 +19,9 @@ def test_create_role_definition(admin_api_client):
     Test creation of a custom role definition.
     """
     url = get_relative_url("roledefinition-list")
-    data = dict(name='foo-role-def', description='bar', permissions=['aap.view_organization', 'aap.change_organization'], content_type='shared.organization')
+    data = dict(
+        name='foo-role-def', description='bar', permissions=['shared.view_organization', 'shared.change_organization'], content_type='shared.organization'
+    )
     response = admin_api_client.post(url, data=data, format="json")
     assert response.status_code == 201, response.data
     assert response.data['name'] == 'foo-role-def'
@@ -28,7 +30,7 @@ def test_create_role_definition(admin_api_client):
 @pytest.mark.django_db
 def test_create_global_role_definition(admin_api_client):
     url = get_relative_url("roledefinition-list")
-    data = dict(name='global_view_org', description='bar', permissions=['aap.view_organization'])
+    data = dict(name='global_view_org', description='bar', permissions=['shared.view_organization'])
     response = admin_api_client.post(url, data=data, format="json")
     assert response.status_code == 201, response.data
     assert response.data['name'] == 'global_view_org'

test_app/tests/rbac/features/test_creator_permission.py

@@ -64,14 +64,14 @@ def test_custom_creation_perms(rando, inventory):
 
 
 @pytest.mark.django_db
-def test_creator_permission_for_unregistered_model(admin_api_client, inventory, inv_rd, user):
+def test_creator_permission_for_unregistered_model(rando):
     prior_ct = DABContentType.objects.count()
     prior_assignments = RoleUserAssignment.objects.count()
     prior_rds = RoleDefinition.objects.count()
 
     cls = apps.get_model('test_app.secretcolor')
     obj = cls.objects.create()
-    RoleDefinition.objects.give_creator_permissions(user, obj)  # should do nothing
+    RoleDefinition.objects.give_creator_permissions(rando, obj)  # should do nothing
 
     assert DABContentType.objects.count() == prior_ct  # did not create anything
     assert RoleUserAssignment.objects.count() == prior_assignments

test_app/tests/rbac/features/test_public_model.py

@@ -45,7 +45,7 @@ def test_org_level_validator_without_view():
 @pytest.mark.django_db
 def test_custom_role_for_public_model(admin_api_client, rando, public_item):
     url = get_relative_url('roledefinition-list')
-    data = {'name': 'Public data editor', 'permissions': ['local.change_publicdata'], 'content_type': 'local.publicdata'}
+    data = {'name': 'Public data editor', 'permissions': ['aap.change_publicdata'], 'content_type': 'aap.publicdata'}
     response = admin_api_client.post(url, data=data, format="json")
     assert response.status_code == 201, response.data
 

test_app/tests/rbac/models/test_content_type.py

@@ -79,16 +79,15 @@ def test_lookup_cache(self):
             DABContentType.objects.get_for_model(Inventory)
 
     @isolate_apps("tests")
-    def test_get_for_model_create_contenttype(self):
+    def test_get_for_model_not_registered(self):
         class ModelCreatedOnTheFly(models.Model):
             name = models.CharField(max_length=10)
 
             class Meta:
                 app_label = "tests"
 
-        ct = DABContentType.objects.get_for_model(ModelCreatedOnTheFly)
-        assert ct.app_label == "tests"
-        assert ct.model == "modelcreatedonthefly"
+        with pytest.raises(RuntimeError):
+            DABContentType.objects.get_for_model(ModelCreatedOnTheFly)
 
 
 @pytest.mark.django_db

test_app/tests/rbac/remote/test_remote_assignment.py

@@ -1,6 +1,6 @@
 import pytest
 
-from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleUserAssignment
+from ansible_base.rbac.models import RoleUserAssignment
 from ansible_base.rbac.remote import RemoteObject
 
 

test_app/tests/rbac/test_validators.py

@@ -30,7 +30,7 @@ def test_custom_role_rules_do_not_apply_to_managed_roles():
 @override_settings(ANSIBLE_BASE_ALLOW_CUSTOM_ROLES=False)
 def test_role_definition_enablement_validation_in_api(admin_api_client):
     url = get_relative_url('roledefinition-list')
-    r = admin_api_client.post(url, data={'name': 'foo', 'permissions': ['view_inventory'], 'content_type': 'aap.inventory'})
+    r = admin_api_client.post(url, data={'name': 'foo', 'permissions': ['aap.view_inventory'], 'content_type': 'aap.inventory'})
     assert r.status_code == 400, r.data
     assert 'Creating custom roles is disabled' in str(r.data)