""" we use middlewares in this project and this file store classes of our middlewares """ from rest_framework import status from authenticate.middleware import AuthorizatorMiddleware from payment.models import UserBalanceHistory from quiz.models import Quiz, QuizSheet, QUIZ_SHEET_STATE_SCORES_PUBLISHED, QUIZ_SHEET_STATE_FINISHED from quiz.serializers import GetQuizSheetMinimumDataForStudentsSerializer, QuizModelSerializer from quiz.utils import price_calculator_helper from utils.myresponse import MyResponse from utils.utils import logs_adder class EntranceValidatorMiddleware: def __init__(self, get_response): self.get_response = get_response # One-time configuration and initialization. def __call__(self, request): # Code to be executed for each request before # the view (and later middleware) are called. response = self.get_response(request) # Code to be executed for each request/response after # the view is called. return response def process_view(self, request, view_func, view_args, view_kwargs): quiz = None quiz_sheet = None user = None institute_student = None entrance_type = None token = request.middleware_serializer_data.get('token') entrance_token = request.middleware_serializer_data.get('entrance_token') quiz_id = request.middleware_serializer_data.get('quiz_id') quiz_sheet_id = request.middleware_serializer_data.get('quiz_sheet_id') # these lines have been added to check the coverage of our tests relating to different patterns of inputs if token: if entrance_token: # not possible, prevented in token_quiz_id_quiz_sheet_id_entrance_token_validator logs_adder(f'CRITICAL: Invalid State #23497422187324') else: if quiz_id: if quiz_sheet_id: # not possible, prevented in token_quiz_id_quiz_sheet_id_entrance_token_validator logs_adder(f'CRITICAL: Invalid State #028944224532') else: # good combination. token and quiz_id i = 0 else: if quiz_sheet_id: # good combination. token and quiz_sheet_id i = 0 else: # not possible, prevented in token_quiz_id_quiz_sheet_id_entrance_token_validator # only token logs_adder(f'CRITICAL: Invalid State #199234617542') else: if entrance_token: if quiz_id: # not possible, prevented in token_quiz_id_quiz_sheet_id_entrance_token_validator logs_adder(f'CRITICAL: Invalid State #39829148520943') else: if quiz_sheet_id: # not possible, prevented in token_quiz_id_quiz_sheet_id_entrance_token_validator logs_adder(f'CRITICAL: Invalid State #84728758287552') else: # good combination. entrance_token i = 0 else: if quiz_id: if quiz_sheet_id: # not possible, prevented in token_quiz_id_quiz_sheet_id_entrance_token_validator logs_adder(f'CRITICAL: Invalid State #3879573229855') else: # good combination. quiz_id i = 0 else: if quiz_sheet_id: # good combination. quiz_id i = 0 else: # not possible, prevented in token_quiz_id_quiz_sheet_id_entrance_token_validator # no value was sent logs_adder(f'CRITICAL: Invalid State #917497527575') if entrance_type is None: try: if entrance_token is not None: quiz_sheet = QuizSheet.objects.get(entrance_token=entrance_token) quiz = quiz_sheet.quiz institute_student = quiz_sheet.institute_student entrance_type = "InstituteStudent" except Exception as e: pass if entrance_type is None: try: quiz = Quiz.objects.get(pk=quiz_id) except Exception as e: pass try: quiz_sheet = QuizSheet.objects.get(pk=quiz_sheet_id) quiz = quiz_sheet.quiz except Exception as e: pass if quiz is not None: user_auth_result = AuthorizatorMiddleware(None, accept_user_token=True).process_view(request, None, None, None) try: user = request.middleware_user except: pass if quiz_sheet is not None and quiz_sheet.user != user: return MyResponse(request, {'status': 'not_found', 'message': 'No quiz sheet found related to your request.'}, status=status.HTTP_404_NOT_FOUND) if quiz.public: entrance_type = "Public" if entrance_type is None: if not user_auth_result: if quiz.institute.user.id != request.middleware_user.id: # TODO handle AdminInvite pass else: entrance_type = "Owner" if entrance_type is None or quiz is None: return MyResponse(request, {'status': 'not_found', 'message': 'No quiz found related to your request.'}, status=status.HTTP_404_NOT_FOUND) request.middleware_entrance_type = entrance_type request.middleware_quiz = quiz request.middleware_user = user request.middleware_institute_student = institute_student request.middleware_quiz_sheet = quiz_sheet class QuizSheetStatusValidationMiddleware: def __init__(self, get_response, error_if_quiz_sheet_is_none=True, process_quiz_status=True): self.get_response = get_response self.error_if_quiz_sheet_is_none = error_if_quiz_sheet_is_none self.process_quiz_status = process_quiz_status # One-time configuration and initialization. def __call__(self, request): # Code to be executed for each request before # the view (and later middleware) are called. response = self.get_response(request) # Code to be executed for each request/response after # the view is called. return response def process_view(self, request, view_func, view_args, view_kwargs): # if request.middleware_quiz.paused: # return MyResponse(request, {'status': 'quiz_paused', 'message': 'The quiz has paused'}, # status=status.HTTP_400_BAD_REQUEST) quiz_sheet = request.middleware_quiz_sheet if quiz_sheet is None: if self.error_if_quiz_sheet_is_none: return MyResponse(request, {'status': 'error', 'message': 'Cannot resolve a quiz_sheet'}, status=status.HTTP_400_BAD_REQUEST) else: # we only process quiz status if quiz sheet is None if self.process_quiz_status: quiz = request.middleware_quiz error_status, message = quiz.can_be_entered_state() if error_status is not None: quiz_data = QuizModelSerializer(quiz, context={"with_blueprint": False}).data return MyResponse(request, {'status': error_status, 'message': message, 'data': {'quiz': quiz_data}}, status=status.HTTP_406_NOT_ACCEPTABLE) return error_status, message = quiz_sheet.can_be_continued_state() if error_status is not None and quiz_sheet.state != QUIZ_SHEET_STATE_SCORES_PUBLISHED and quiz_sheet.state != QUIZ_SHEET_STATE_FINISHED: from quiz.views import quiz_sheet_finish_helper quiz_sheet_finish_helper(quiz_sheet) if error_status is not None: quiz_sheet_data = GetQuizSheetMinimumDataForStudentsSerializer(quiz_sheet).data return MyResponse(request, {'status': error_status, 'message': message, 'data': {'quiz_sheet': quiz_sheet_data}}, status=status.HTTP_406_NOT_ACCEPTABLE) class QuizPurchaseHandlerMiddleware: def __init__(self, get_response): self.get_response = get_response # One-time configuration and initialization. def __call__(self, request): # Code to be executed for each request before # the view (and later middleware) are called. response = self.get_response(request) # Code to be executed for each request/response after # the view is called. return response def process_view(self, request, view_func, view_args, view_kwargs): if request.middleware_user is None and request.middleware_institute_student is None: return MyResponse(request, {'status': 'error', 'message': 'Access denied'}, status=status.HTTP_401_UNAUTHORIZED) if request.middleware_quiz_sheet is None: total_cost, institute_share, wikiazma_share, wage, tax = price_calculator_helper(request.middleware_user, request.middleware_institute_student, request.middleware_quiz) user_balance = UserBalanceHistory.get_last_user_balance(request.middleware_user).remaining_balance if user_balance < total_cost: return MyResponse(request, {'status': 'error', 'message': 'balance is not enough', 'data': {'total_cost': total_cost, 'balance': user_balance}}, status=status.HTTP_402_PAYMENT_REQUIRED)