# By: Riasat Ullah # This file contains views for handling heartbeat configurations. from constants import api_paths, label_names as lnm, pages, static_vars, var_names from constants import component_names as cnm from context_manager import automation_context from django.core import exceptions from django.http import JsonResponse from django.shortcuts import redirect, render from django.views.decorators.http import require_http_methods from taskcallweb import settings from system_tests.test_data import test_data_checks from translators import label_translator as lt from utils import logging, helpers from validations import request_validator import json @require_http_methods(['GET', 'POST']) def get_heartbeat_list(request): ''' GET: Gets the page that lists the heartbeat checks. POST: Gets the list of heartbeat checks. :param request: Http request :return: GET: Http response | POST: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): lang = request_validator.get_user_language(request) nav_components = request_validator.get_nav_bar_components(request) has_view_perm, has_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_heartbeats, nav_components ) if not has_view_perm: raise exceptions.PermissionDenied context = automation_context.get_heartbeat_list_context(lang, nav_components) context[var_names.has_edit_permission] = request_validator.has_edit_permission( request, cnm.dis_com_heartbeats) context[var_names.has_team_permission] = request_validator.has_edit_permission( request, cnm.dis_com_teams) return render(request, pages.heartbeat_list_page, context=context) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse(test_data_checks.heartbeats_list, safe=False) else: status, output = helpers.post_api_request(api_paths.checks_heartbeat_list, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def get_heartbeat_details(request, check_ref_id=None): ''' GET: Gets the heartbeat details page. POST: Gets all the details for a specific heartbeat. :param request: Http request :param check_ref_id: the reference ID of the heartbeat check; None implies new heartbeat check :return: GET: Http response | POST: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): try: lang = request_validator.get_user_language(request) nav_components = request_validator.get_nav_bar_components(request) has_view_perm, has_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_heartbeats, nav_components ) if not has_view_perm: raise exceptions.PermissionDenied context = automation_context.get_heartbeat_details_context(lang, nav_components) context[var_names.context] = json.dumps({var_names.check_ref_id: check_ref_id}) context[var_names.has_edit_permission] = has_edit_perm context[var_names.has_team_permission] = request_validator.has_edit_permission( request, cnm.dis_com_teams) context[var_names.has_contextual_search_permission] = request_validator.has_edit_permission( request, cnm.dis_com_tags) return render(request, pages.heartbeat_details_page, context=context) except ValueError as e: logging.exception(str(e)) return render(request, pages.heartbeat_list_page) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: for item in test_data_checks.heartbeats_list: if item[var_names.check_ref_id] == check_ref_id: return JsonResponse(item, safe=False) return JsonResponse(None, status=404, safe=False) else: body = {var_names.check_ref_id: check_ref_id} status, output = helpers.post_api_request(api_paths.checks_heartbeat_details, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def get_heartbeat_metrics(request, check_ref_id): ''' GET: Gets the heartbeat metrics details page. POST: Gets the metrics of a heartbeat. :param request: Http request :param check_ref_id: the reference ID of the heartbeat; if none is provided, then it means it is for a new rule :return: GET: Http response | POST: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): try: lang = request_validator.get_user_language(request) nav_components = request_validator.get_nav_bar_components(request) has_view_perm, has_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_heartbeats, nav_components ) if not has_view_perm: raise exceptions.PermissionDenied context = automation_context.get_heartbeat_metrics_context(lang, nav_components) context[var_names.context] = json.dumps({var_names.check_ref_id: check_ref_id}) context[var_names.has_edit_permission] = has_edit_perm return render(request, pages.heartbeat_metrics_page, context=context) except ValueError as e: logging.exception(str(e)) return render(request, pages.heartbeat_list_page) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse(test_data_checks.heartbeats_metrics[check_ref_id], safe=False) else: ip_address = request.META.get(static_vars.ip_address_attribute) timezone = helpers.get_info_from_ip_address(ip_address, static_vars.ip_timezone_attribute) body[var_names.timezone] = timezone status, output = helpers.post_api_request(api_paths.checks_heartbeat_metrics, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=401, safe=False) @require_http_methods(['POST']) def get_heartbeat_endpoints(request): ''' Get the potential url/email endpoints for a heartbeat check. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: output = { var_names.check_ref_id: test_data_checks.heartbeats_list[0][var_names.check_ref_id], var_names.url: test_data_checks.heartbeats_list[0][var_names.url], var_names.email: test_data_checks.heartbeats_list[0][var_names.email] } return JsonResponse(output, safe=False) else: status, output = helpers.post_api_request( api_paths.checks_heartbeat_endpoints, dict(), request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False)