implement more natural natural keys (#16514)

src/authentic2/a2_rbac/models.py

@@ -93,6 +93,9 @@ class OrganizationalUnit(OrganizationalUnitAbstractBase):
         return cls.objects.all()
 
 
+OrganizationalUnit._meta.natural_key = [['uuid'], ['slug'], ['name']]
+
+
 class Permission(PermissionAbstractBase):
     class Meta:
         verbose_name = _('permission')
@@ -103,6 +106,9 @@ class Permission(PermissionAbstractBase):
                                    object_id_field='admin_scope_id')
 
 
+Permission._meta.natural_key = ['operation', 'ou', 'target']
+
+
 class Role(RoleAbstractBase):
     admin_scope_ct = models.ForeignKey(
         to='contenttypes.ContentType',
@@ -208,6 +214,11 @@ class Role(RoleAbstractBase):
         }
 
 
+Role._meta.natural_key = [
+    ['uuid'], ['slug', 'ou'], ['name', 'ou'], ['slug', 'service'], ['name', 'service']
+]
+
+
 class RoleParenting(RoleParentingAbstractBase):
     class Meta(RoleParentingAbstractBase.Meta):
         verbose_name = _('role parenting relation')

src/authentic2/models.py

@@ -23,6 +23,8 @@ except ImportError:
 from django.contrib.contenttypes.models import ContentType
 
 from . import managers
+# install our natural_key implementation
+from . import natural_key
 from .utils import ServiceAccessDenied
 
 
@@ -407,6 +409,9 @@ class Service(models.Model):
         }
 
 
+Service._meta.natural_key = [['slug', 'ou']]
+
+
 class AuthorizedRole(models.Model):
     service = models.ForeignKey(Service, on_delete=models.CASCADE)
     role = models.ForeignKey(get_role_model_name(), on_delete=models.CASCADE)

src/authentic2/natural_key.py

@@ -0,0 +1,99 @@
+from django.db import models
+
+from django.contrib.contenttypes.models import ContentType
+from django.contrib.contenttypes.fields import GenericForeignKey
+
+
+def get_natural_keys(model):
+    if not getattr(model._meta, 'natural_key', None):
+        raise ValueError('model %s has no natural key defined in its Meta' % model.__name__)
+    natural_key = model._meta.natural_key
+    if not hasattr(natural_key, '__iter__'):
+        raise ValueError('natural_key must be an iterable')
+    if hasattr(natural_key[0], 'lower'):
+        natural_key = [natural_key]
+    return natural_key
+
+
+def natural_key_json(self):
+    natural_keys = get_natural_keys(self.__class__)
+    d = {}
+    names = set()
+    for keys in natural_keys:
+        for key in keys:
+            names.add(key)
+
+    for name in names:
+        field = self._meta.get_field(name)
+        if not (field.concrete or isinstance(field, GenericForeignKey)):
+            raise ValueError('field %s is not concrete' % name)
+        if field.is_relation and not field.many_to_one:
+            raise ValueError('field %s is a relation but not a ForeignKey' % name)
+        value = getattr(self, name)
+        if isinstance(field, GenericForeignKey):
+            ct_field_value = getattr(self, field.ct_field)
+            d[field.ct_field] = ct_field_value and ct_field_value.natural_key_json()
+            d[name] = value and value.natural_key_json()
+        elif field.is_relation:
+            d[name] = value and value.natural_key_json()
+        else:
+            d[name] = value
+    return d
+
+
+def get_by_natural_key_json(self, d):
+    model = self.model
+    natural_keys = get_natural_keys(model)
+    if not isinstance(d, dict):
+        raise ValueError('a natural_key must be a dictionnary')
+    for natural_key in natural_keys:
+        get_kwargs = {}
+        for name in natural_key:
+            field = model._meta.get_field(name)
+            if not (field.concrete or isinstance(field, GenericForeignKey)):
+                raise ValueError('field %s is not concrete' % name)
+            if field.is_relation and not field.many_to_one:
+                raise ValueError('field %s is a relation but not a ForeignKey' % name)
+            try:
+                value = d[name]
+            except KeyError:
+                break
+            if isinstance(field, GenericForeignKey):
+                try:
+                    ct_nk = d[field.ct_field]
+                except KeyError:
+                    break
+                try:
+                    ct = ContentType.objects.get_by_natural_key_json(ct_nk)
+                except ContentType.DoesNotExist:
+                    break
+                related_model = ct.model_class()
+                try:
+                    value = related_model._default_manager.get_by_natural_key_json(value)
+                except related_model.DoesNotExist:
+                    break
+                get_kwargs[field.ct_field] = ct
+                name = field.fk_field
+                value = value.pk
+            elif field.is_relation:
+                if value is None:
+                    name = '%s__isnull' % name
+                    value = True
+                else:
+                    try:
+                        value = field.related_model._default_manager.get_by_natural_key_json(value)
+                    except field.related_model.DoesNotExist:
+                        break
+            get_kwargs[name] = value
+        else:
+            try:
+                return self.get(**get_kwargs)
+            except model.DoesNotExist:
+                pass
+    raise model.DoesNotExist
+
+
+models.Model.natural_key_json = natural_key_json
+models.Manager.get_by_natural_key_json = get_by_natural_key_json
+
+ContentType._meta.natural_key = ['app_label', 'model']

src/django_rbac/models.py

@@ -117,6 +117,9 @@ class Operation(models.Model):
     objects = managers.OperationManager()
 
 
+Operation._meta.natural_key = ['slug']
+
+
 class PermissionAbstractBase(models.Model):
     operation = models.ForeignKey(
         to='Operation',

tests/test_natural_key.py

@@ -0,0 +1,52 @@
+from django.contrib.contenttypes.models import ContentType
+from authentic2.a2_rbac.models import Role, OrganizationalUnit as OU, Permission
+
+
+def test_natural_key_json(db, ou1):
+    role = Role.objects.create(slug='role1', name='Role1', ou=ou1)
+
+    for ou in OU.objects.all():
+        nk = ou.natural_key_json()
+        assert nk == {'uuid': ou.uuid, 'slug': ou.slug, 'name': ou.name}
+
+        assert ou == OU.objects.get_by_natural_key_json(nk)
+
+    for ct in ContentType.objects.all():
+        nk = ct.natural_key_json()
+        assert nk == {'app_label': ct.app_label, 'model': ct.model}
+        assert ct == ContentType.objects.get_by_natural_key_json(nk)
+
+    # test is not useful if there are no FK set
+    assert Role.objects.filter(ou__isnull=False).exists()
+
+    for role in Role.objects.all():
+        nk = role.natural_key_json()
+        ou_nk = role.ou and role.ou.natural_key_json()
+        service_nk = role.service and role.service.natural_key_json()
+        assert nk == {
+            'uuid': role.uuid, 'slug': role.slug, 'name': role.name, 'ou': ou_nk,
+            'service': service_nk
+        }
+        assert role == Role.objects.get_by_natural_key_json(nk)
+        assert role == Role.objects.get_by_natural_key_json({'uuid': role.uuid})
+        assert role == Role.objects.get_by_natural_key_json({'slug': role.slug, 'ou': ou_nk})
+        assert role == Role.objects.get_by_natural_key_json({'name': role.name, 'ou': ou_nk})
+        assert role == Role.objects.get_by_natural_key_json(
+            {'slug': role.slug, 'service': service_nk})
+        assert role == Role.objects.get_by_natural_key_json(
+            {'name': role.name, 'service': service_nk})
+
+    for permission in Permission.objects.all():
+        ou_nk = permission.ou and permission.ou.natural_key_json()
+        target_ct_nk = permission.target_ct.natural_key_json()
+        target_nk = permission.target.natural_key_json()
+        op_nk = permission.operation.natural_key_json()
+
+        nk = permission.natural_key_json()
+        assert nk == {
+            'operation': op_nk,
+            'ou': ou_nk,
+            'target_ct': target_ct_nk,
+            'target': target_nk,
+        }
+        assert permission == Permission.objects.get_by_natural_key_json(nk)