Coverage for backend_app/models/abstract/essentialModule.py : 87%

from django.contrib.contenttypes.fields import GenericRelation 

from django.contrib.contenttypes.models import ContentType 

from django.core.exceptions import ValidationError 

from django.core.validators import MinValueValidator 

from django.db import models 

from django.utils import timezone 

from rest_framework import serializers 

from backend_app.models.pendingModeration import PendingModeration 

from backend_app.permissions.app_permissions import NoPost, NoDelete 

from backend_app.permissions.moderation import is_moderation_required 

from backend_app.permissions.utils import Request as FakeRequest 

from backend_app.settings.defaults import ( 

DEFAULT_OBJ_MODERATION_LV, 

OBJ_MODERATION_PERMISSIONS, 

) 

from backend_app.utils import get_user_level 

from base_app.models import User 

from .base import BaseModel, BaseModelSerializer, BaseModelViewSet 

POSSIBLE_OBJ_MODER_LV = [ 

OBJ_MODERATION_PERMISSIONS[key] for key in OBJ_MODERATION_PERMISSIONS 

] 

def validate_obj_model_lv(value): 

if value not in POSSIBLE_OBJ_MODER_LV: 

raise ValidationError("obj_moderation_level not recognized") 

# 

# 

# 

# 

# 

# 

# Module 

# 

# 

# 

# 

# 

# 

# 

# 

class EssentialModule(BaseModel): 

""" 

All models in the app depend of this one. 

It contains the required attributes for managing optional data moderation. 

All the logic behind moderation is done in EssentialModuleSerializer 

""" 

# store the update author 

updated_by = models.ForeignKey( 

User, null=True, on_delete=models.SET_NULL, related_name="+" 

) 

# store the update date (model can be updated without moderation) 

updated_on = models.DateTimeField(null=True) 

# store the moderator 

moderated_by = models.ForeignKey( 

User, null=True, on_delete=models.SET_NULL, related_name="+" 

) 

# store the moderation date 

moderated_on = models.DateTimeField(null=True) 

# Store the object moderation level by default 

obj_moderation_level = models.SmallIntegerField( 

default=DEFAULT_OBJ_MODERATION_LV, 

validators=[MinValueValidator(0), validate_obj_model_lv], 

) 

# Add the link to pending moderation 

pending_moderation = GenericRelation(PendingModeration) 

# A bit of optimization: we store if there is something pending moderation 

has_pending_moderation = models.BooleanField(default=False) 

class Meta: 

abstract = True 

# 

# 

# 

# 

# 

# 

# Serializer 

# 

# 

# 

# 

# 

# 

# 

# 

CLEANED_ESSENTIAL_MODULE_MODEL_DATA = { 

"moderated_by": None, 

"moderated_on": None, 

"updated_by": None, 

"updated_on": None, 

} 

def override_data(old_data: dict, new_data: dict) -> dict: 

"""Update the data in old_data with the one in new_data 

""" 

for key in new_data: 

if key in old_data: 

old_data[key] = new_data[key] 

return old_data 

class EssentialModuleSerializer(BaseModelSerializer): 

"""Serializer to go along the EssentialModule Model. This serializer handles backend data moderation checks and tricks. 

Raises: 

ValidationError -- If you are trying to moderate something you don't have rights to 

""" 

check_obj_permissions_for_edit = False 

###### 

# Basic fields serializers 

updated_by = serializers.SerializerMethodField() 

updated_on = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S", read_only=True) 

moderated_by = serializers.SerializerMethodField() 

moderated_on = serializers.DateTimeField(format="%Y-%m-%d %H:%M:%S", read_only=True) 

has_pending_moderation = serializers.BooleanField(read_only=True) 

# Add a content_type_id field to be able to find versions 

content_type_id = serializers.SerializerMethodField() 

def get_updated_by(self, obj): 

return self.get_user_related_field(obj.updated_by) 

def get_moderated_by(self, obj): 

return self.get_user_related_field(obj.moderated_by) 

def get_content_type_id(self, obj): 

""" 

Serializer for content type 

""" 

return ContentType.objects.get_for_model(self.Meta.model).id 

def get_obj_info(self, obj) -> dict: 

""" 

Serializer for the `obj_info` *dynamic* field. Redefined. 

""" 

obj_info = super().get_obj_info(obj) 

try: 

user_can_edit = self.context["user_can_edit"] 

except KeyError: 

# In case some viewsets don't inherit from BaseModelViewSet and therefore 

# don't have the method to produce context["user_can_edit"] 

# Anyway, those Viewsets should be readonly, so we can return false. 

user_can_edit = False 

if user_can_edit and self.check_obj_permissions_for_edit: 

try: 

fake_edit_request = FakeRequest(self.get_user_from_request(), "PUT") 

for permission_class in self.context["permission_classes"]: 

if not permission_class.has_object_permission( 

fake_edit_request, None, obj 

): 

user_can_edit = False 

break 

except KeyError: 

pass 

obj_info["user_can_edit"] = user_can_edit 

obj_info["user_can_moderate"] = user_can_edit and not is_moderation_required( 

self.Meta.model, obj, self.get_user_from_request() 

) 

return obj_info 

class Meta: 

model = EssentialModule 

fields = BaseModelSerializer.Meta.fields + ( 

"updated_by", 

"updated_on", 

"moderated_by", 

"moderated_on", 

"has_pending_moderation", 

"content_type_id", 

) 

def validate(self, attrs): 

""" 

Validate `BaseModel` fields and enforce certain field at the backend level. 

Checks that the requested moderation level is not higher than the one of the user. 

""" 

if "obj_moderation_level" in attrs: 

requested_obj_moder_lv = attrs["obj_moderation_level"] 

try: 

user_level = get_user_level(self.get_user_from_request()) 

except KeyError: 

# if for some reason we don't have the user in the request 

# we set the level to the default one 

# this can occur during testing. 

user_level = DEFAULT_OBJ_MODERATION_LV 

if requested_obj_moder_lv > user_level: 

raise ValidationError( 

"You can't request moderation for a higher rank than you." 

) 

return attrs 

def set_model_attrs_for_moderation_and_update( 

self, user, moderated_and_updated: bool 

): 

""" 

Overrides model attributes regarding moderation and update. 

The moderated field is set to the request user. The moderated_on field is reset to now. 

If there was an updated, the updated_by field and updated_on field are also reset. 

""" 

now = timezone.now() 

self.override_validated_data({"moderated_by": user, "moderated_on": now}) 

if moderated_and_updated: 

self.override_validated_data({"updated_by": user, "updated_on": now}) 

def clean_validated_data(self): 

""" 

Clear fields related to update and moderation 

""" 

self.override_validated_data(CLEANED_ESSENTIAL_MODULE_MODEL_DATA) 

def override_validated_data(self, new_data: dict): 

""" 

Method used to force specific attributes when saving a model 

""" 

for key in new_data: 

self.validated_data[key] = new_data[key] 

def do_before_save(self): 

""" 

Action to perform before saving a model 

""" 

pass 

def save(self, *args, **kwargs): 

""" 

Function that handles all the moderation in a smart way. 

Nothing has to be done to tell that we won't the data to be moderated, 

it is detected automatically. 

""" 

user = self.context["request"].user 

user_level = get_user_level(user) 

self.clean_validated_data() 

self.do_before_save() 

ct = ContentType.objects.get_for_model(self.Meta.model) 

if is_moderation_required(self.Meta.model, self.instance, user, user_level): 

if self.instance is None: # we need to create the main model 

# Store the user for squashing data in versions models 

self.validated_data.updated_by = user 

self.instance = super().save(*args, **kwargs) 

data_to_save = dict() 

for key in self.validated_data: 

try: 

# retrieve the submitted data and save the clean json 

# to make sure it will be savable 

data_to_save[key] = self.initial_data[key] 

except KeyError: 

pass 

data_to_save = override_data( 

data_to_save, CLEANED_ESSENTIAL_MODULE_MODEL_DATA 

) 

# Save instance into pending moderation state 

PendingModeration.objects.update_or_create( 

content_type=ct, 

object_id=self.instance.pk, 

defaults={ 

"updated_on": timezone.now(), 

"updated_by": user, 

"new_object": data_to_save, 

}, 

) 

# Performance optimization, we store the fact that there is an object pending moderation 

self.instance.has_pending_moderation = True 

self.instance.save() 

return self.instance 

else: # Moderation is not needed, we need to check whether it's a moderation or an update with no moderation 

moderated_and_updated = True 

if self.instance is None: 

self.set_model_attrs_for_moderation_and_update( 

user, moderated_and_updated 

) 

instance = super().save(*args, **kwargs) 

else: 

try: 

pending_instance = PendingModeration.objects.get( 

content_type=ct, object_id=self.instance.pk 

) 

self.clean_validated_data() # Make quez that it is done... 

# We have to compare the serialized data 

# So we make sure to compare the same elements 

key_to_remove = [] 

for key in self.initial_data: 

if key not in self.validated_data: 

key_to_remove.append(key) 

for key in key_to_remove: 

self.initial_data.pop(key, None) 

if pending_instance.new_object == self.initial_data: 

moderated_and_updated = False 

self.validated_data["updated_by"] = pending_instance.updated_by 

self.validated_data["updated_on"] = pending_instance.updated_on 

pending_instance.delete() 

except PendingModeration.DoesNotExist: 

pass 

self.set_model_attrs_for_moderation_and_update( 

user, moderated_and_updated 

) 

instance = super().save(*args, **kwargs) 

# Performance optimization to know if has pending moderation 

instance.has_pending_moderation = False 

instance.save() 

return instance 

# 

# 

# 

# 

# 

# 

# ViewSet 

# 

# 

# 

# 

# 

# 

# 

# 

class EssentialModuleViewSet(BaseModelViewSet): 

""" 

Custom default viewset 

""" 

permission_classes = (NoPost & NoDelete,) 

serializer_class = EssentialModuleSerializer 

def get_serializer_context(self): 

""" 

Override default function. 

Extra context is provided to the serializer class to 

know if a user can edit an element or not. 

This allows to not do this query for all elements and improves 

performances. You can look at the comment below for more information. 

""" 

# When generating the API documentation (url: /api-doc) the request would be None 

# and we don't need to do anything special 

if self.request is None: 

return super().get_serializer_context() 

fake_edit_request = FakeRequest(self.request.user, "PUT") 

user_can_edit = True 

for permission_class in self.get_permissions(): 

# Theoretically speaking we would need to use has_object_permission 

# But for performance purpose, we will consider edition right at the model 

# level. Which is consistent with our design. 

# Beware, that this might provide inconsistent data to the frontend 

# especially if permission_classes impact at the object level such as 

# IsOwner. 

# 

# Set check_obj_permissions_for_edit=True in your serializer 

# if you want a better check at the object level 

if not permission_class.has_permission(fake_edit_request, None): 

user_can_edit = False 

break 

default_context = super().get_serializer_context() 

default_context["user_can_edit"] = user_can_edit 

default_context["permission_classes"] = self.get_permissions() 

return default_context 

def get_queryset(self): 

""" 

Extended default rest framework behavior 

to prefetch some table and enhance performances 

""" 

return self.queryset.prefetch_related("moderated_by", "updated_by")